IntenseFinance/flectra
3
0
Fork 0
mirror of https://gitlab.com/flectra-hq/flectra.git synced 2026-08-17 16:54:42 -05:00
Code Issues Packages Projects Releases Wiki Activity

[PATCH] Upstream patch - 25122023

Browse Source
This commit is contained in:
Parthiv Patel
2023-12-25 08:34:55 +00:00
parent 905908812e
commit 2c20de7b4d
2 changed files with 44 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
addons/auth_signup/models/res_partner.py
+19 -3
View File
@@ -8,7 +8,7 @@ from collections import defaultdict
from datetime import datetime, timedelta
from flectra import api, exceptions, fields, models, _
from flectra.tools import sql
class SignupError(Exception):
pass
@@ -24,12 +24,17 @@ def now(**kwargs):
class ResPartner(models.Model):
_inherit = 'res.partner'
signup_token = fields.Char(copy=False, groups="base.group_erp_manager")
signup_token = fields.Char(copy=False, groups="base.group_erp_manager", compute='_compute_token', inverse='_inverse_token')
signup_type = fields.Char(string='Signup Token Type', copy=False, groups="base.group_erp_manager")
signup_expiration = fields.Datetime(copy=False, groups="base.group_erp_manager")
signup_valid = fields.Boolean(compute='_compute_signup_valid', string='Signup Token is Valid')
signup_url = fields.Char(compute='_compute_signup_url', string='Signup URL')
def init(self):
super().init()
if not sql.column_exists(self.env.cr, self._table, "signup_token"):
self.env.cr.execute("ALTER TABLE res_partner ADD COLUMN signup_token varchar")
@api.depends('signup_token', 'signup_expiration')
def _compute_signup_valid(self):
dt = now()
@@ -47,6 +52,15 @@ class ResPartner(models.Model):
self.env['res.partner'].check_access_rights('write')
partner.signup_url = result.get(partner.id, False)
def _compute_token(self):
for partner in self.filtered('id'):
self.env.cr.execute('SELECT signup_token FROM res_partner WHERE id=%s', (partner._origin.id,))
partner.signup_token = self.env.cr.fetchone()[0]
def _inverse_token(self):
for partner in self.filtered('id'):
self.env.cr.execute('UPDATE res_partner SET signup_token = %s WHERE id=%s', (partner.signup_token or None, partner.id))
def _get_signup_url_for_action(self, url=None, action=None, view_type=None, menu_id=None, res_id=None, model=None):
""" generate a signup url for the given partner ids and action, possibly overriding
the url state components (menu_id, id, view_type) """
@@ -145,7 +159,9 @@ class ResPartner(models.Model):
:param raise_exception: if True, raise exception instead of returning False
:return: partner (browse record) or False (if raise_exception is False)
"""
partner = self.search([('signup_token', '=', token)], limit=1)
self.env.cr.execute("SELECT id FROM res_partner WHERE signup_token = %s AND active", (token,))
partner_id = self.env.cr.fetchone()
partner = self.browse(partner_id[0]) if partner_id else None
if not partner:
if raise_exception:
raise exceptions.UserError(_("Signup token '%s' is not valid", token))
addons/auth_totp/models/res_users.py
+25 -4
View File
@@ -15,6 +15,7 @@ from flectra import _, api, fields, models
from flectra.addons.base.models.res_users import check_identity
from flectra.exceptions import AccessDenied, UserError
from flectra.http import request, db_list
from flectra.tools import sql
_logger = logging.getLogger(__name__)
@@ -24,14 +25,16 @@ compress = functools.partial(re.sub, r'\s', '')
class Users(models.Model):
_inherit = 'res.users'
totp_secret = fields.Char(copy=False, groups=fields.NO_ACCESS)
totp_enabled = fields.Boolean(string="Two-factor authentication", compute='_compute_totp_enabled')
totp_secret = fields.Char(copy=False, groups=fields.NO_ACCESS, compute='_compute_totp_secret', inverse='_inverse_totp_secret')
totp_enabled = fields.Boolean(string="Two-factor authentication", compute='_compute_totp_enabled', search='_search_totp_enable')
totp_trusted_device_ids = fields.One2many('res.users.apikeys', 'user_id',
string="Trusted Devices", domain=[('scope', '=', TRUSTED_DEVICE_SCOPE)])
api_key_ids = fields.One2many(domain=[('scope', '!=', TRUSTED_DEVICE_SCOPE)])
def __init__(self, pool, cr):
init_res = super().__init__(pool, cr)
if not sql.column_exists(cr, self._table, "totp_secret"):
cr.execute("ALTER TABLE res_users ADD COLUMN totp_secret varchar")
type(self).SELF_READABLE_FIELDS = self.SELF_READABLE_FIELDS + ['totp_enabled', 'totp_trusted_device_ids']
return init_res
@@ -67,9 +70,9 @@ class Users(models.Model):
key = base64.b32decode(sudo.totp_secret)
match = TOTP(key).match(code)
if match is None:
_logger.info("2FA check: FAIL for %s %r", self, self.login)
_logger.info("2FA check: FAIL for %s %r", self, sudo.login)
raise AccessDenied()
_logger.info("2FA check: SUCCESS for %s %r", self, self.login)
_logger.info("2FA check: SUCCESS for %s %r", self, sudo.login)
def _totp_try_setting(self, secret, code):
if self.totp_enabled or self != self.env.user:
@@ -156,6 +159,24 @@ class Users(models.Model):
self.env.user._revoke_all_devices()
return super().change_password(old_passwd, new_passwd)
def _compute_totp_secret(self):
for user in self.filtered('id'):
self.env.cr.execute('SELECT totp_secret FROM res_users WHERE id=%s', (user.id,))
user.totp_secret = self.env.cr.fetchone()[0]
def _inverse_totp_secret(self):
for user in self.filtered('id'):
secret = user.totp_secret if user.totp_secret else None
self.env.cr.execute('UPDATE res_users SET totp_secret = %s WHERE id=%s', (secret, user.id))
def _search_totp_enable(self, operator, value):
value = not value if operator == '!=' else value
if value:
self.env.cr.execute("SELECT id FROM res_users WHERE totp_secret IS NOT NULL")
else:
self.env.cr.execute("SELECT id FROM res_users WHERE totp_secret IS NULL OR totp_secret='false'")
result = self.env.cr.fetchall()
return [('id', 'in', [x[0] for x in result])]
class TOTPWizard(models.TransientModel):
_name = 'auth_totp.wizard'