engineering/unrhodecode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(2fa): refactor logic arround validation/recoverycodes and workflows of configuration of 2fa

Browse source 
- recovery codes are shown in 1 place only
- save status about view of recovery codes
- made the logic of saving states into user_data more explicit and no longer relly on hacky DB transaction logic
- turn JS forms into a regular forms
This commit is contained in:
RhodeCode Admin 2024-04-24 09:45:36 +02:00
parent 2fcf5e8c05
commit 337031c7a4
11 changed files with 353 additions and 270 deletions
Download patch file Download diff file Expand all files Collapse all files

127
rhodecode/model/db.py
View file

@ -796,34 +796,13 @@ class User(Base, BaseModel):
Session.commit()
return artifact_token.api_key
@hybrid_property
def secret_2fa(self):
if not self.user_data.get('secret_2fa'):
secret = pyotp.random_base32()
self.update_userdata(secret_2fa=safe_str(enc_utils.encrypt_value(secret, enc_key=ENCRYPTION_KEY)))
return secret
return safe_str(
enc_utils.decrypt_value(self.user_data['secret_2fa'],
enc_key=ENCRYPTION_KEY,
strict_mode=ConfigGet().get_bool('rhodecode.encrypted_values.strict',
missing=True)
)
)
def is_totp_valid(self, received_code):
totp = pyotp.TOTP(self.secret_2fa)
def is_totp_valid(self, received_code, secret):
totp = pyotp.TOTP(secret)
return totp.verify(received_code)
def is_2fa_recovery_code_valid(self, received_code):
def is_2fa_recovery_code_valid(self, received_code, secret):
encrypted_recovery_codes = self.user_data.get('recovery_codes_2fa', [])
recovery_codes = list(map(
lambda x: safe_str(
enc_utils.decrypt_value(
x,
enc_key=ENCRYPTION_KEY,
strict_mode=ConfigGet().get_bool('rhodecode.encrypted_values.strict', missing=True)
)),
encrypted_recovery_codes))
recovery_codes = self.get_2fa_recovery_codes()
if received_code in recovery_codes:
encrypted_recovery_codes.pop(recovery_codes.index(received_code))
self.update_userdata(recovery_codes_2fa=encrypted_recovery_codes)
@ -844,7 +823,7 @@ class User(Base, BaseModel):
@hybrid_property
def has_enabled_2fa(self):
"""
Checks if 2fa was enabled by user
Checks if user enabled 2fa
"""
if value := self.has_forced_2fa:
return value
@ -853,34 +832,109 @@ class User(Base, BaseModel):
@has_enabled_2fa.setter
def has_enabled_2fa(self, val):
val = str2bool(val)
self.update_userdata(enabled_2fa=str2bool(val))
self.update_userdata(enabled_2fa=val)
if not val:
self.update_userdata(secret_2fa=None, recovery_codes_2fa=[])
# NOTE: setting to false we clear the user_data to not store any 2fa artifacts
self.update_userdata(secret_2fa=None, recovery_codes_2fa=[], check_2fa=False)
Session().commit()
def get_2fa_recovery_codes(self):
@hybrid_property
def has_check_2fa_flag(self):
"""
Check if check 2fa flag is set for this user
"""
value = self.user_data.get('check_2fa', False)
return value
@has_check_2fa_flag.setter
def has_check_2fa_flag(self, val):
val = str2bool(val)
self.update_userdata(check_2fa=val)
Session().commit()
@hybrid_property
def has_seen_2fa_codes(self):
"""
get the flag about if user has seen 2fa recovery codes
"""
value = self.user_data.get('recovery_codes_2fa_seen', False)
return value
@has_seen_2fa_codes.setter
def has_seen_2fa_codes(self, val):
val = str2bool(val)
self.update_userdata(recovery_codes_2fa_seen=val)
Session().commit()
@hybrid_property
def needs_2fa_configure(self):
"""
Determines if setup2fa has completed for this user. Means he has all needed data for 2fa to work.
Currently this is 2fa enabled and secret exists
"""
if self.has_enabled_2fa:
return not self.user_data.get('secret_2fa')
return False
def init_2fa_recovery_codes(self, persist=True, force=False):
"""
Creates 2fa recovery codes
"""
recovery_codes = self.user_data.get('recovery_codes_2fa', [])
encrypted_codes = []
if not recovery_codes:
if not recovery_codes or force:
for _ in range(self.RECOVERY_CODES_COUNT):
recovery_code = pyotp.random_base32()
recovery_codes.append(recovery_code)
encrypted_codes.append(safe_str(enc_utils.encrypt_value(recovery_code, enc_key=ENCRYPTION_KEY)))
self.update_userdata(recovery_codes_2fa=encrypted_codes)
encrypted_code = enc_utils.encrypt_value(safe_bytes(recovery_code), enc_key=ENCRYPTION_KEY)
encrypted_codes.append(safe_str(encrypted_code))
if persist:
self.update_userdata(recovery_codes_2fa=encrypted_codes, recovery_codes_2fa_seen=False)
return recovery_codes
# User should not check the same recovery codes more than once
return []
def get_2fa_recovery_codes(self):
encrypted_recovery_codes = self.user_data.get('recovery_codes_2fa', [])
strict_mode = ConfigGet().get_bool('rhodecode.encrypted_values.strict', missing=True)
recovery_codes = list(map(
lambda val: safe_str(
enc_utils.decrypt_value(
val,
enc_key=ENCRYPTION_KEY,
strict_mode=strict_mode
)),
encrypted_recovery_codes))
return recovery_codes
def init_secret_2fa(self, persist=True, force=False):
secret_2fa = self.user_data.get('secret_2fa')
if not secret_2fa or force:
secret = pyotp.random_base32()
if persist:
self.update_userdata(secret_2fa=safe_str(enc_utils.encrypt_value(safe_bytes(secret), enc_key=ENCRYPTION_KEY)))
return secret
return ''
def get_secret_2fa(self) -> str:
secret_2fa = self.user_data['secret_2fa']
if secret_2fa:
strict_mode = ConfigGet().get_bool('rhodecode.encrypted_values.strict', missing=True)
return safe_str(
enc_utils.decrypt_value(secret_2fa, enc_key=ENCRYPTION_KEY, strict_mode=strict_mode))
return ''
def set_2fa_secret(self, value):
encrypted_value = enc_utils.encrypt_value(safe_bytes(value), enc_key=ENCRYPTION_KEY)
self.update_userdata(secret_2fa=safe_str(encrypted_value))
def regenerate_2fa_recovery_codes(self):
"""
Regenerates 2fa recovery codes upon request
"""
self.update_userdata(recovery_codes_2fa=[])
Session().flush()
new_recovery_codes = self.get_2fa_recovery_codes()
new_recovery_codes = self.init_2fa_recovery_codes(force=True)
Session().commit()
return new_recovery_codes
@ -5021,8 +5075,7 @@ class Gist(Base, BaseModel):
return data
def __json__(self):
data = dict(
)
data = dict()
data.update(self.get_api_data())
return data
# SCM functions