fix(2fa): fixed typo from refactor

rhodecode/apps/login/views.py

@@ -1,553 +1,553 b''
 # Copyright (C) 2016-2023 RhodeCode GmbH
 #
 # This program is free software: you can redistribute it and/or modify
 # it under the terms of the GNU Affero General Public License, version 3
 # (only), as published by the Free Software Foundation.
 #
 # This program is distributed in the hope that it will be useful,
 # but WITHOUT ANY WARRANTY; without even the implied warranty of
 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 # GNU General Public License for more details.
 #
 # You should have received a copy of the GNU Affero General Public License
 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
 #
 # This program is dual-licensed. If you wish to learn more about the
 # RhodeCode Enterprise Edition, including its added features, Support services,
 # and proprietary license terms, please see https://rhodecode.com/licenses/
 
 import time
 import json
 import pyotp
 import qrcode
 import collections
 import datetime
 import formencode
 import formencode.htmlfill
 import logging
 import urllib.parse
 import requests
 from io import BytesIO
 from base64 import b64encode
 
 from pyramid.renderers import render
 from pyramid.response import Response
 from pyramid.httpexceptions import HTTPFound
 
 import rhodecode
 from rhodecode.apps._base import BaseAppView
 from rhodecode.authentication.base import authenticate, HTTP_TYPE
 from rhodecode.authentication.plugins import auth_rhodecode
 from rhodecode.events import UserRegistered, trigger
 from rhodecode.lib import helpers as h
 from rhodecode.lib import audit_logger
 from rhodecode.lib.auth import (
     AuthUser, HasPermissionAnyDecorator, CSRFRequired, LoginRequired, NotAnonymous)
 from rhodecode.lib.base import get_ip_addr
 from rhodecode.lib.exceptions import UserCreationError
 from rhodecode.lib.utils2 import safe_str
 from rhodecode.model.db import User, UserApiKeys
 from rhodecode.model.forms import LoginForm, RegisterForm, PasswordResetForm, TOTPForm
 from rhodecode.model.meta import Session
 from rhodecode.model.auth_token import AuthTokenModel
 from rhodecode.model.settings import SettingsModel
 from rhodecode.model.user import UserModel
 from rhodecode.translation import _
 
 
 log = logging.getLogger(__name__)
 
 CaptchaData = collections.namedtuple(
     'CaptchaData', 'active, private_key, public_key')
 
 
 def store_user_in_session(session, user_identifier, remember=False):
     user = User.get_by_username_or_primary_email(user_identifier)
     auth_user = AuthUser(user.user_id)
     auth_user.set_authenticated()
     cs = auth_user.get_cookie_store()
     session['rhodecode_user'] = cs
     user.update_lastlogin()
     Session().commit()
 
     # If they want to be remembered, update the cookie
     if remember:
         _year = (datetime.datetime.now() +
                  datetime.timedelta(seconds=60 * 60 * 24 * 365))
         session._set_cookie_expires(_year)
 
     session.save()
 
     safe_cs = cs.copy()
     safe_cs['password'] = '****'
     log.info('user %s is now authenticated and stored in '
              'session, session attrs %s', user_identifier, safe_cs)
 
     # dumps session attrs back to cookie
     session._update_cookie_out()
     # we set new cookie
     headers = None
     if session.request['set_cookie']:
         # send set-cookie headers back to response to update cookie
         headers = [('Set-Cookie', session.request['cookie_out'])]
     return headers
 
 
 def get_came_from(request):
     came_from = safe_str(request.GET.get('came_from', ''))
     parsed = urllib.parse.urlparse(came_from)
 
     allowed_schemes = ['http', 'https']
     default_came_from = h.route_path('home')
     if parsed.scheme and parsed.scheme not in allowed_schemes:
         log.error('Suspicious URL scheme detected %s for url %s',
                   parsed.scheme, parsed)
         came_from = default_came_from
     elif parsed.netloc and request.host != parsed.netloc:
         log.error('Suspicious NETLOC detected %s for url %s server url '
                   'is: %s', parsed.netloc, parsed, request.host)
         came_from = default_came_from
     elif any(bad_char in came_from for bad_char in ('\r', '\n')):
         log.error('Header injection detected `%s` for url %s server url ',
                   parsed.path, parsed)
         came_from = default_came_from
 
     return came_from or default_came_from
 
 
 class LoginView(BaseAppView):
 
     def load_default_context(self):
         c = self._get_local_tmpl_context()
         c.came_from = get_came_from(self.request)
         return c
 
     def _get_captcha_data(self):
         settings = SettingsModel().get_all_settings()
         private_key = settings.get('rhodecode_captcha_private_key')
         public_key = settings.get('rhodecode_captcha_public_key')
         active = bool(private_key)
         return CaptchaData(
             active=active, private_key=private_key, public_key=public_key)
 
     def validate_captcha(self, private_key):
 
         captcha_rs = self.request.POST.get('g-recaptcha-response')
         url = "https://www.google.com/recaptcha/api/siteverify"
         params = {
             'secret': private_key,
             'response': captcha_rs,
             'remoteip': get_ip_addr(self.request.environ)
         }
         verify_rs = requests.get(url, params=params, verify=True, timeout=60)
         verify_rs = verify_rs.json()
         captcha_status = verify_rs.get('success', False)
         captcha_errors = verify_rs.get('error-codes', [])
         if not isinstance(captcha_errors, list):
             captcha_errors = [captcha_errors]
         captcha_errors = ', '.join(captcha_errors)
         captcha_message = ''
         if captcha_status is False:
             captcha_message = "Bad captcha. Errors: {}".format(
                 captcha_errors)
 
         return captcha_status, captcha_message
 
     def login(self):
         c = self.load_default_context()
         auth_user = self._rhodecode_user
 
         # redirect if already logged in
         if (auth_user.is_authenticated and
                 not auth_user.is_default and auth_user.ip_allowed):
             raise HTTPFound(c.came_from)
 
         # check if we use headers plugin, and try to login using it.
         try:
             log.debug('Running PRE-AUTH for headers based authentication')
             auth_info = authenticate(
                 '', '', self.request.environ, HTTP_TYPE, skip_missing=True)
             if auth_info:
                 headers = store_user_in_session(
                     self.session, auth_info.get('username'))
                 raise HTTPFound(c.came_from, headers=headers)
         except UserCreationError as e:
             log.error(e)
             h.flash(e, category='error')
 
         return self._get_template_context(c)
 
     def login_post(self):
         c = self.load_default_context()
 
         login_form = LoginForm(self.request.translate)()
 
         try:
             self.session.invalidate()
             form_result = login_form.to_python(self.request.POST)
             # form checks for username/password, now we're authenticated
             username = form_result['username']
             if (user := User.get_by_username_or_primary_email(username)).has_enabled_2fa:
                 user.check_2fa_required = True
 
             headers = store_user_in_session(
                 self.session,
                 user_identifier=username,
                 remember=form_result['remember'])
             log.debug('Redirecting to "%s" after login.', c.came_from)
 
             audit_user = audit_logger.UserWrap(
                 username=self.request.POST.get('username'),
                 ip_addr=self.request.remote_addr)
             action_data = {'user_agent': self.request.user_agent}
             audit_logger.store_web(
                 'user.login.success', action_data=action_data,
                 user=audit_user, commit=True)
 
             raise HTTPFound(c.came_from, headers=headers)
         except formencode.Invalid as errors:
             defaults = errors.value
             # remove password from filling in form again
             defaults.pop('password', None)
             render_ctx = {
                 'errors': errors.error_dict,
                 'defaults': defaults,
             }
 
             audit_user = audit_logger.UserWrap(
                 username=self.request.POST.get('username'),
                 ip_addr=self.request.remote_addr)
             action_data = {'user_agent': self.request.user_agent}
             audit_logger.store_web(
                 'user.login.failure', action_data=action_data,
                 user=audit_user, commit=True)
             return self._get_template_context(c, **render_ctx)
 
         except UserCreationError as e:
             # headers auth or other auth functions that create users on
             # the fly can throw this exception signaling that there's issue
             # with user creation, explanation should be provided in
             # Exception itself
             h.flash(e, category='error')
             return self._get_template_context(c)
 
     @CSRFRequired()
     def logout(self):
         auth_user = self._rhodecode_user
         log.info('Deleting session for user: `%s`', auth_user)
 
         action_data = {'user_agent': self.request.user_agent}
         audit_logger.store_web(
             'user.logout', action_data=action_data,
             user=auth_user, commit=True)
         self.session.delete()
         return HTTPFound(h.route_path('home'))
 
     @HasPermissionAnyDecorator(
         'hg.admin', 'hg.register.auto_activate', 'hg.register.manual_activate')
     def register(self, defaults=None, errors=None):
         c = self.load_default_context()
         defaults = defaults or {}
         errors = errors or {}
 
         settings = SettingsModel().get_all_settings()
         register_message = settings.get('rhodecode_register_message') or ''
         captcha = self._get_captcha_data()
         auto_active = 'hg.register.auto_activate' in User.get_default_user()\
             .AuthUser().permissions['global']
 
         render_ctx = self._get_template_context(c)
         render_ctx.update({
             'defaults': defaults,
             'errors': errors,
             'auto_active': auto_active,
             'captcha_active': captcha.active,
             'captcha_public_key': captcha.public_key,
             'register_message': register_message,
         })
         return render_ctx
 
     @HasPermissionAnyDecorator(
         'hg.admin', 'hg.register.auto_activate', 'hg.register.manual_activate')
     def register_post(self):
         from rhodecode.authentication.plugins import auth_rhodecode
 
         self.load_default_context()
         captcha = self._get_captcha_data()
         auto_active = 'hg.register.auto_activate' in User.get_default_user()\
             .AuthUser().permissions['global']
 
         extern_name = auth_rhodecode.RhodeCodeAuthPlugin.uid
         extern_type = auth_rhodecode.RhodeCodeAuthPlugin.uid
 
         register_form = RegisterForm(self.request.translate)()
         try:
 
             form_result = register_form.to_python(self.request.POST)
             form_result['active'] = auto_active
             external_identity = self.request.POST.get('external_identity')
 
             if external_identity:
                 extern_name = external_identity
                 extern_type = external_identity
 
             if captcha.active:
                 captcha_status, captcha_message = self.validate_captcha(
                     captcha.private_key)
 
                 if not captcha_status:
                     _value = form_result
                     _msg = _('Bad captcha')
                     error_dict = {'recaptcha_field': captcha_message}
                     raise formencode.Invalid(
                         _msg, _value, None, error_dict=error_dict)
 
             new_user = UserModel().create_registration(
                 form_result, extern_name=extern_name, extern_type=extern_type)
 
             action_data = {'data': new_user.get_api_data(),
                            'user_agent': self.request.user_agent}
 
             if external_identity:
                 action_data['external_identity'] = external_identity
 
             audit_user = audit_logger.UserWrap(
                 username=new_user.username,
                 user_id=new_user.user_id,
                 ip_addr=self.request.remote_addr)
 
             audit_logger.store_web(
                 'user.register', action_data=action_data,
                 user=audit_user)
 
             event = UserRegistered(user=new_user, session=self.session)
             trigger(event)
             h.flash(
                 _('You have successfully registered with RhodeCode. You can log-in now.'),
                 category='success')
             if external_identity:
                 h.flash(
                     _('Please use the {identity} button to log-in').format(
                         identity=external_identity),
                     category='success')
             Session().commit()
 
             redirect_ro = self.request.route_path('login')
             raise HTTPFound(redirect_ro)
 
         except formencode.Invalid as errors:
             errors.value.pop('password', None)
             errors.value.pop('password_confirmation', None)
             return self.register(
                 defaults=errors.value, errors=errors.error_dict)
 
         except UserCreationError as e:
             # container auth or other auth functions that create users on
             # the fly can throw this exception signaling that there's issue
             # with user creation, explanation should be provided in
             # Exception itself
             h.flash(e, category='error')
             return self.register()
 
     def password_reset(self):
         c = self.load_default_context()
         captcha = self._get_captcha_data()
 
         template_context = {
             'captcha_active': captcha.active,
             'captcha_public_key': captcha.public_key,
             'defaults': {},
             'errors': {},
         }
 
         # always send implicit message to prevent from discovery of
         # matching emails
         msg = _('If such email exists, a password reset link was sent to it.')
 
         def default_response():
             log.debug('faking response on invalid password reset')
             # make this take 2s, to prevent brute forcing.
             time.sleep(2)
             h.flash(msg, category='success')
             return HTTPFound(self.request.route_path('reset_password'))
 
         if self.request.POST:
             if h.HasPermissionAny('hg.password_reset.disabled')():
                 _email = self.request.POST.get('email', '')
                 log.error('Failed attempt to reset password for `%s`.', _email)
                 h.flash(_('Password reset has been disabled.'), category='error')
                 return HTTPFound(self.request.route_path('reset_password'))
 
             password_reset_form = PasswordResetForm(self.request.translate)()
             description = 'Generated token for password reset from {}'.format(
                 datetime.datetime.now().isoformat())
 
             try:
                 form_result = password_reset_form.to_python(
                     self.request.POST)
                 user_email = form_result['email']
 
                 if captcha.active:
                     captcha_status, captcha_message = self.validate_captcha(
                         captcha.private_key)
 
                     if not captcha_status:
                         _value = form_result
                         _msg = _('Bad captcha')
                         error_dict = {'recaptcha_field': captcha_message}
                         raise formencode.Invalid(
                             _msg, _value, None, error_dict=error_dict)
 
                 # Generate reset URL and send mail.
                 user = User.get_by_email(user_email)
 
                 # only allow rhodecode based users to reset their password
                 # external auth shouldn't allow password reset
                 if user and user.extern_type != auth_rhodecode.RhodeCodeAuthPlugin.uid:
                     log.warning('User %s with external type `%s` tried a password reset. '
                                 'This try was rejected', user, user.extern_type)
                     return default_response()
 
                 # generate password reset token that expires in 10 minutes
                 reset_token = UserModel().add_auth_token(
                     user=user, lifetime_minutes=10,
                     role=UserModel.auth_token_role.ROLE_PASSWORD_RESET,
                     description=description)
                 Session().commit()
 
                 log.debug('Successfully created password recovery token')
                 password_reset_url = self.request.route_url(
                     'reset_password_confirmation',
                     _query={'key': reset_token.api_key})
                 UserModel().reset_password_link(
                     form_result, password_reset_url)
 
                 action_data = {'email': user_email,
                                'user_agent': self.request.user_agent}
                 audit_logger.store_web(
                     'user.password.reset_request', action_data=action_data,
                     user=self._rhodecode_user, commit=True)
 
                 return default_response()
 
             except formencode.Invalid as errors:
                 template_context.update({
                     'defaults': errors.value,
                     'errors': errors.error_dict,
                 })
                 if not self.request.POST.get('email'):
                     # case of empty email, we want to report that
                     return self._get_template_context(c, **template_context)
 
                 if 'recaptcha_field' in errors.error_dict:
                     # case of failed captcha
                     return self._get_template_context(c, **template_context)
 
             return default_response()
 
         return self._get_template_context(c, **template_context)
 
     def password_reset_confirmation(self):
         self.load_default_context()
 
         if key := self.request.GET.get('key'):
             # make this take 2s, to prevent brute forcing.
             time.sleep(2)
 
             token = AuthTokenModel().get_auth_token(key)
 
             # verify token is the correct role
             if token is None or token.role != UserApiKeys.ROLE_PASSWORD_RESET:
                 log.debug('Got token with role:%s expected is %s',
                           getattr(token, 'role', 'EMPTY_TOKEN'),
                           UserApiKeys.ROLE_PASSWORD_RESET)
                 h.flash(
                     _('Given reset token is invalid'), category='error')
                 return HTTPFound(self.request.route_path('reset_password'))
 
             try:
                 owner = token.user
                 data = {'email': owner.email, 'token': token.api_key}
                 UserModel().reset_password(data)
                 h.flash(
                     _('Your password reset was successful, '
                       'a new password has been sent to your email'),
                     category='success')
             except Exception as e:
                 log.error(e)
                 return HTTPFound(self.request.route_path('reset_password'))
 
         return HTTPFound(self.request.route_path('login'))
 
     @LoginRequired()
     @NotAnonymous()
     def setup_2fa(self):
         _ = self.request.translate
         c = self.load_default_context()
         user_instance = self._rhodecode_db_user
         form = TOTPForm(_, user_instance)()
         render_ctx = {}
         if self.request.method == 'POST':
             post_items = dict(self.request.POST)
 
             try:
                 form_details = form.to_python(post_items)
                 secret = form_details['secret_totp']
 
                 user_instance.init_2fa_recovery_codes(persist=True, force=True)
-                user_instance.2fa_secret = secret
+                user_instance.secret_2fa = secret
 
                 Session().commit()
                 raise HTTPFound(self.request.route_path('my_account_configure_2fa', _query={'show-recovery-codes': 1}))
             except formencode.Invalid as errors:
                 defaults = errors.value
                 render_ctx = {
                     'errors': errors.error_dict,
                     'defaults': defaults,
                 }
 
         # NOTE: here we DO NOT persist the secret 2FA, since this is only for setup, once a setup is completed
         # only then we should persist it
         secret = user_instance.init_secret_2fa(persist=False)
 
         instance_name = rhodecode.ConfigGet().get_str('app.base_url', 'rhodecode')
         totp_name = f'{instance_name}:{self.request.user.username}'
 
         qr = qrcode.QRCode(version=1, box_size=5, border=4)
         qr.add_data(pyotp.totp.TOTP(secret).provisioning_uri(name=totp_name))
         qr.make(fit=True)
         img = qr.make_image(fill_color='black', back_color='white')
         buffered = BytesIO()
         img.save(buffered)
         return self._get_template_context(
             c,
             qr=b64encode(buffered.getvalue()).decode("utf-8"),
             key=secret,
             totp_name=totp_name,
             ** render_ctx
         )
 
     @LoginRequired()
     @NotAnonymous()
     def verify_2fa(self):
         _ = self.request.translate
         c = self.load_default_context()
         render_ctx = {}
         user_instance = self._rhodecode_db_user
         totp_form = TOTPForm(_, user_instance, allow_recovery_code_use=True)()
         if self.request.method == 'POST':
             post_items = dict(self.request.POST)
             # NOTE: inject secret, as it's a post configured saved item.
             post_items['secret_totp'] = user_instance.secret_2fa
             try:
                 totp_form.to_python(post_items)
                 user_instance.check_2fa_required = False
                 Session().commit()
                 raise HTTPFound(c.came_from)
             except formencode.Invalid as errors:
                 defaults = errors.value
                 render_ctx = {
                     'errors': errors.error_dict,
                     'defaults': defaults,
                 }
         return self._get_template_context(c, **render_ctx)