[Feature] TOTP 2FA support

js/log-in-link-is-valid.js

@@ -0,0 +1,12 @@
+// Helper to append an optional TOTP-code to the login link when signing in via email
+window.addEventListener("load", function () {
+    var totp_field = document.getElementById("log-in-link-is-valid_totp_field");
+    if (totp_field) {
+        var login_button = document.getElementById("log-in-link-is-valid_button");
+            totp_field.addEventListener("input", function () {
+            var url = new URLSearchParams(login_button.href);
+            url.set("log-in.totp", totp_field.value);
+            login_button.href = decodeURIComponent(url.toString());
+        });
+    }
+});

liberapay/constants.py

@@ -437,4 +437,6 @@ def __missing__(self, currency):
 USERNAME_MAX_SIZE = 32
 USERNAME_SUFFIX_BLACKLIST = set('.txt .html .htm .json .xml'.split())
 
+VALID_WINDOW_2FA = 5
+
 del _

liberapay/exceptions.py

@@ -77,6 +77,10 @@ class AccountIsPasswordless(LoginRequired):
     pass
 
 
+class FailedToVerifyOTP(Exception):
+    pass
+
+
 class NeedDatabase(LazyResponse):
     html_template = 'templates/exceptions/NeedDatabase.html'
 

liberapay/models/participant.py

@@ -12,6 +12,9 @@
 import unicodedata
 from urllib.parse import quote as urlquote, urlencode
 import uuid
+import io
+from pyotp import totp, random_base32
+from pyqrcode import create as create_qrcode
 
 import aspen_jinja2_renderer
 from cached_property import cached_property
@@ -32,7 +35,7 @@
     PASSWORD_MAX_SIZE, PASSWORD_MIN_SIZE, PAYPAL_CURRENCIES,
     PERIOD_CONVERSION_RATES, PRIVILEGES,
     PUBLIC_NAME_MAX_SIZE, SEPA, SESSION, SESSION_TIMEOUT,
-    USERNAME_MAX_SIZE, USERNAME_SUFFIX_BLACKLIST,
+    USERNAME_MAX_SIZE, USERNAME_SUFFIX_BLACKLIST, VALID_WINDOW_2FA
 )
 from liberapay.exceptions import (
     AccountIsPasswordless,
@@ -45,6 +48,7 @@
     EmailAddressIsBlacklisted,
     EmailAlreadyTaken,
     EmailNotVerified,
+    FailedToVerifyOTP,
     InvalidId,
     LoginRequired,
     NonexistingElsewhere,
@@ -240,12 +244,13 @@ def refetch(self):
     # ===================
 
     @classmethod
-    def authenticate_with_password(cls, p_id, password, context='log-in'):
+    def authenticate_with_password(cls, p_id, password, totp='', context='log-in'):
         """Fetch a participant using its ID, but only if the provided password is valid.
 
         Args:
             p_id (int): the participant's ID
             password (str): the participant's password
+            totp (str): optional totp code
             context (str): the operation that this authentication is part of
 
         Return type: `Participant | None`
@@ -266,6 +271,9 @@ def authenticate_with_password(cls, p_id, password, context='log-in'):
             return None
         p, stored_secret = r
         if context == 'log-in':
+            if p.is_totp_enabled():
+                if (not totp) or (not p.verify_totp(totp)):
+                    return None
             cls.db.hit_rate_limit('log-in.password', p.id, TooManyPasswordLogins)
         request = website.state.get({}).get('request')
         if request:
@@ -388,27 +396,82 @@ def check_password(self, password, context):
             self.add_event(website.db, 'password-check', None)
         return status
 
+    # 2FA Management (TOTP)
+    # =====================
+
+    def is_totp_enabled(self):
+        return self.db.one(
+            "SELECT totp_verified FROM participants WHERE id = %s",
+            (self.id,)
+        )
+
+    def gen_totp_token(self):
+        totp_token = self.db.one(
+            "SELECT totp_token FROM participants WHERE id = %s",
+            (self.id,)
+        )
+        if totp_token is None:
+            totp_token = random_base32()
+            self.db.run(
+                "UPDATE participants SET totp_token = %s WHERE id = %s",
+                (totp_token, self.id)
+            )
+        return totp_token
+
+    def generate_totp_qrcode(self):
+        totp_token = self.gen_totp_token()
+        uri = totp.TOTP(totp_token).provisioning_uri(
+            name=f'~{self.id}',
+            image='https://liberapay.com/assets/liberapay/icon-v2_white-on-yellow.200.png',
+            issuer_name='Liberapay'
+        )
+        qrcode = create_qrcode(uri)
+        buffer = io.BytesIO()
+        qrcode.png(buffer, scale=4)
+        image = b64encode(buffer.getvalue()).decode()
+        return image
+
+    def verify_totp(self, totp_code):
+        totp_token = self.gen_totp_token()
+        return totp.TOTP(totp_token).verify(totp_code, valid_window=VALID_WINDOW_2FA)
+
+    def enable_totp(self, verification_code):
+        if self.verify_totp(verification_code):
+            self.db.run("UPDATE participants SET totp_verified = true WHERE id = %s", (self.id,))
+        else:
+            raise FailedToVerifyOTP
+
+    def disable_totp(self):
+        self.db.run("""
+            UPDATE participants
+                SET totp_token = NULL,
+                    totp_verified = false
+            WHERE id = %s
+            """, (self.id,))
+
 
     # Session Management
     # ==================
 
     @classmethod
     def authenticate_with_session(
-        cls, p_id, session_id, secret, allow_downgrade=False, cookies=None,
+        cls, p_id, session_id, secret, totp='', allow_downgrade=False, cookies=None, context='log-in'
     ):
         """Fetch a participant using its ID, but only if the provided session is valid.
 
         Args:
             p_id (int | str): the participant's ID
             session_id (int | str): the ID of the session
             secret (str): the actual secret
+            totp (str): optional totp code
             allow_downgrade (bool): allow downgrading to a read-only session
             cookies (SimpleCookie):
                 the response cookies, only needed when `allow_downgrade` is `True`
+            context (str): the operation that this authentication is part of
 
         Return type:
             Tuple[Participant, Literal['valid']] |
-            Tuple[None, Literal['expired', 'invalid']]
+            Tuple[None, Literal['expired', 'invalid', 'require_totp']]
         """
         if not secret:
             if session_id == '!':
@@ -431,6 +494,10 @@ def authenticate_with_session(
         if not r:
             return None, 'invalid'
         p, stored_secret, mtime = r
+        if context == 'log-in':
+            if p.is_totp_enabled():
+                if (not totp) or (not p.verify_totp(totp)):
+                    return p, 'require_totp'
         if not constant_time_compare(stored_secret, secret):
             return None, 'invalid'
         if mtime > utcnow() - SESSION_TIMEOUT:

liberapay/security/authentication.py

@@ -70,6 +70,7 @@ def sign_in_with_form_data(body, state):
         src_addr, src_country = request.source, request.country
         input_id = body['log-in.id'].strip()
         password = body.pop('log-in.password', None)
+        totp = body.pop('log-in.totp', None)
         id_type = None
         if input_id.find('@') > 0:
             id_type = 'email'
@@ -84,7 +85,7 @@ def sign_in_with_form_data(body, state):
             else:
                 p_id = Participant.get_id_for(id_type, input_id)
             try:
-                p = Participant.authenticate_with_password(p_id, password)
+                p = Participant.authenticate_with_password(p_id, password, totp=totp)
             except AccountIsPasswordless:
                 if id_type == 'email':
                     state['log-in.email'] = input_id
@@ -93,7 +94,7 @@ def sign_in_with_form_data(body, state):
                 return
             if not p:
                 state['log-in.error'] = (
-                    _("The submitted password is incorrect.") if p_id is not None else
+                    _("The submitted password and/or otp is incorrect.") if p_id is not None else
                     _("“{0}” is not a valid account ID.", input_id) if id_type == 'immutable' else
                     _("No account has the username “{username}”.", username=input_id) if id_type == 'username' else
                     _("No account has “{email_address}” as its primary email address.", email_address=input_id)
@@ -261,7 +262,7 @@ def authenticate_user_if_possible(csrf_token, request, response, state, user, _)
             creds = [creds[0], 1, creds[1]]
         if len(creds) == 3:
             session_p, state['session_status'] = Participant.authenticate_with_session(
-                *creds, allow_downgrade=True, cookies=response.headers.cookie
+                *creds, allow_downgrade=True, cookies=response.headers.cookie, context='cookie'
             )
             if session_p:
                 user = state['user'] = session_p
@@ -307,19 +308,20 @@ def authenticate_user_if_possible(csrf_token, request, response, state, user, _)
         if request.qs.get('log-in.id'):
             # Email auth
             id = request.qs.get_int('log-in.id')
+            totp = request.qs.get('log-in.totp')
             session_id = request.qs.get('log-in.key')
             if not session_id or session_id < '1001' or session_id > '1010':
                 raise response.render('simplates/log-in-link-is-invalid.spt', state)
             token = request.qs.get('log-in.token')
             if not (token and token.endswith('.em')):
                 raise response.render('simplates/log-in-link-is-invalid.spt', state)
             required = request.qs.parse_boolean('log-in.required', default=True)
-            p = Participant.authenticate_with_session(
+            p, reason = Participant.authenticate_with_session(
                 id, session_id, token,
-                allow_downgrade=not required, cookies=response.headers.cookie,
-            )[0]
+                totp=totp, allow_downgrade=not required, cookies=response.headers.cookie
+            )
             if p:
-                if p.id != user.id:
+                if p.id != user.id or reason == 'require_totp':
                     submitted_confirmation_token = request.qs.get('log-in.confirmation')
                     if submitted_confirmation_token:
                         expected_confirmation_token = b64encode_s(blake2b(
@@ -338,6 +340,8 @@ def authenticate_user_if_possible(csrf_token, request, response, state, user, _)
                                 'querystring'
                             )
                         del request.qs['log-in.confirmation']
+                        if not p.session and reason == 'require_totp':
+                            raise response.render('simplates/log-in-link-is-valid.spt', state)
                     else:
                         raise response.render('simplates/log-in-link-is-valid.spt', state)
                 redirect = True

requirements_base.txt

@@ -259,3 +259,12 @@ OpenCC==1.1.3 \
 gunicorn==20.1.0 \
     --hash=sha256:9dcc4547dbb1cb284accfb15ab5667a0e5d1881cc443e0677b4882a4067a807e \
     --hash=sha256:e0a968b5ba15f8a328fdfd7ab1fcb5af4470c28aaf7e55df02a99bc13138e6e8
+pyotp==2.7.0 \
+    --hash=sha256:2e746de4f15685878df6d022c5691627af9941eec18e0d513f05497f5fa7711f \
+    --hash=sha256:ce989faba0df77dc032b45e51c6cca42bcf20896c8d3d1e7cd759a53dc7d6cb5
+PyQRCode==1.2.1 \
+    --hash=sha256:1b2812775fa6ff5c527977c4cd2ccb07051ca7d0bc0aecf937a43864abe5eff6 \
+    --hash=sha256:fdbf7634733e56b72e27f9bce46e4550b75a3a2c420414035cae9d9d26b234d5
+pypng==0.20220715.0 \
+    --hash=sha256:4a43e969b8f5aaafb2a415536c1a8ec7e341cd6a3f957fd5b5f32a4cfeed902c \
+    --hash=sha256:739c433ba96f078315de54c0db975aee537cbc3e1d0ae4ed9aab0ca1e427e2c1

simplates/log-in-link-is-valid.spt

@@ -3,6 +3,7 @@ from hashlib import blake2b
 from liberapay.utils import b64encode_s
 
 [---]
+
 cancel = request.qs.parse_boolean('log-in.cancel', default=False)
 p_id = request.qs.get_int('log-in.id')
 if cancel:
@@ -37,7 +38,10 @@ response.code = 200
         ) }}</p>
         <p class="flex-row space-between wrap">
             <span class="flex-col">
-                <a class="btn btn-success" href="{{ request.qs.derive(**{'log-in.confirmation': confirmation_token}) }}">{{ _(
+                <input id="log-in-link-is-valid_totp_field" class="form-control"
+                   type="text"
+                   placeholder="{{ _('One-time password (optional)') }}" />
+                <a id="log-in-link-is-valid_button" class="btn btn-success" href="{{ request.qs.derive(**{'log-in.confirmation': confirmation_token}) }}">{{ _(
                     "Log in as {identifier}",
                     identifier=logging_in_as
                 ) }}</a>

sql/branch.sql

@@ -0,0 +1,5 @@
+-- add columns for TOTP-support in participants
+
+ALTER TABLE participants
+    ADD COLUMN totp_token TEXT,
+    ADD COLUMN totp_verified BOOLEAN DEFAULT false;

templates/log-in-form.html

@@ -44,6 +44,9 @@
             <input name="log-in.password" class="form-control"
                    type="password" autocomplete="current-password"
                    placeholder="{{ _('Password') }}" required />
+            <input name="log-in.totp" class="form-control"
+                   type="text"
+                   placeholder="{{ _('One-time password (optional)') }}" />
         </div>
         <button class="btn btn-primary">{{ _("Log in") }}</button>
     </form>
@@ -91,6 +94,9 @@
 <input name="log-in.password" class="form-control"
        type="password" autocomplete="current-password"
        placeholder="{{ _('Password (optional)') }}" />
+<input name="log-in.totp" class="form-control"
+       type="text"
+       placeholder="{{ _('One-time password (optional)') }}" />
 </div>
 
 <p class="help-block">{{ _(

templates/macros/auth.html

@@ -41,6 +41,14 @@ <h3>{{ _("Password") }}</h3>
     </form>
     <p class="help-block">{{ _("Maximum length is {0}.", constants.PASSWORD_MAX_SIZE) }}</p>
 
-    <h3>{{ _("2FA") }}</h3>
-    <p>{{ _("Liberapay does not yet support two-factor authentication.") }}</p>
+    <h3>{{ _("Two-Factor Authentication (2FA)") }}</h3>
+    % if participant.is_totp_enabled()
+         <form action="{{ participant.path('settings/handle_2fa') }}" method="POST" class="form-inline buttons">
+            <input name="csrf_token" type="hidden" value="{{ csrf_token }}" />
+            <input name="back_to" type="hidden" value="{{ participant.path('settings/') }}" />
+            <button class="btn btn-default" name="action" value="disable">{{ _("Disable Two-Factor Authentication") }}</button>
+        </form>
+    % else
+        <a class="btn btn-default" href="{{ participant.path('settings/enable_2fa') }}">{{ _("Enable Two-Factor Authentication") }}</a>
+    % endif
 % endmacro

tests/py/test_participant.py

@@ -16,12 +16,15 @@
     UsernameIsEmpty,
     UsernameIsRestricted,
     UsernameTooLong,
+    FailedToVerifyOTP,
 )
 from liberapay.i18n.currencies import Money
 from liberapay.models.participant import NeedConfirmation, Participant
 from liberapay.models.tip import Tip
 from liberapay.testing import EUR, USD, Harness
 
+from pyotp import totp
+
 
 class TestTakeOver(Harness):
 
@@ -190,6 +193,44 @@ def test_getting_tips_not_made(self):
         actual = self.stub.get_tip_to(user2).amount
         assert actual == expected
 
+    def test_generate_a_totp_token(self):
+        token = self.stub.gen_totp_token()
+        assert len(token) == 32
+        assert self.stub.is_totp_enabled() == False
+
+    def test_make_sure_that_one_totp_is_generated(self):
+        token = self.stub.gen_totp_token()
+        token2 = self.stub.gen_totp_token()
+        assert token == token2
+        assert self.stub.is_totp_enabled() == False
+
+    def test_generate_new_totp_token(self):
+        token = self.stub.gen_totp_token()
+        self.stub.disable_totp()
+        token2 = self.stub.gen_totp_token()
+        assert token != token2
+        assert self.stub.is_totp_enabled() == False
+
+    def test_verify_totp_successful(self):
+        token = self.stub.gen_totp_token()
+        code = totp.TOTP(token).now()
+        assert self.stub.verify_totp(code) == True
+        assert self.stub.is_totp_enabled() == False
+
+    def test_verify_totp_failure(self):
+        assert self.stub.verify_totp('blah') == False
+        assert self.stub.is_totp_enabled() == False
+
+    def test_enable_totp_successful(self):
+        token = self.stub.gen_totp_token()
+        code = totp.TOTP(token).now()
+        assert self.stub.enable_totp(code) == None
+        assert self.stub.is_totp_enabled() == True
+
+    def test_enable_totp_failure(self):
+        with self.assertRaises(FailedToVerifyOTP):
+            self.stub.enable_totp('blah')
+
 
 class Tests(Harness):