burmatovan/openuds
1
0
Fork 0
forked from shaba/openuds
Code Issues Pull Requests Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/v3.5-mfa'

Browse Source
This commit is contained in:
Adolfo Gómez García 2022-07-02 00:18:09 +02:00
commit 2736390f95
4 changed files with 38 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
server/src/uds/core/mfas/mfa.py
View File

@ -154,7 +154,8 @@ class MFA(Module):
If raises an error, the MFA code was not sent, and the user needs to enter the MFA code.
"""
# try to get the stored code
data: typing.Any = self.storage.getPickle(userId)
storageKey = request.ip + userId
data: typing.Any = self.storage.getPickle(storageKey)
validity = validity if validity is not None else self.validity() * 60
try:
if data and validity:
@ -164,18 +165,18 @@ class MFA(Module):
return MFA.RESULT.OK
except Exception:
# if we have a problem, just remove the stored code
self.storage.remove(userId)
self.storage.remove(storageKey)
# Generate a 6 digit code (0-9)
code = ''.join(random.SystemRandom().choices('0123456789', k=6))
logger.debug('Generated OTP is %s', code)
# Store the code in the database, own storage space
self.storage.putPickle(userId, (getSqlDatetime(), code))
self.storage.putPickle(storageKey, (getSqlDatetime(), code))
# Send the code to the user
return self.sendCode(request, userId, username, identifier, code)
def validate(self, userId: str, username: str, identifier: str, code: str, validity: typing.Optional[int] = None) -> None:
def validate(self, request: 'ExtendedHttpRequest', userId: str, username: str, identifier: str, code: str, validity: typing.Optional[int] = None) -> None:
"""
If this method is provided by an authenticator, the user will be allowed to enter a MFA code
You must raise an "exceptions.MFAError" if the code is not valid.
@ -184,7 +185,8 @@ class MFA(Module):
try:
err = _('Invalid MFA code')
data = self.storage.getPickle(userId)
storageKey = request.ip + userId
data = self.storage.getPickle(storageKey)
if data and len(data) == 2:
validity = validity if validity is not None else self.validity() * 60
if validity and data[0] + datetime.timedelta(seconds=validity) > getSqlDatetime():
@ -194,7 +196,7 @@ class MFA(Module):
# Check if the code is valid
if data[1] == code:
# Code is valid, remove it from storage
self.storage.remove(userId)
self.storage.remove(storageKey)
return
except Exception as e:
# Any error means invalid code

11
server/src/uds/mfas/Email/mfa.py
View File

@ -14,6 +14,7 @@ from uds.core.util import validators, decorators
if typing.TYPE_CHECKING:
from uds.core.module import Module
from uds.core.util.request import ExtendedHttpRequest
logger = logging.getLogger(__name__)
@ -128,7 +129,7 @@ class EmailMFA(mfas.MFA):
return 'OTP received via email'
@decorators.threaded
def doSendCode(self, identifier: str, code: str) -> None:
def doSendCode(self, request: 'ExtendedHttpRequest', identifier: str, code: str) -> None:
# Send and email with the notification
with self.login() as smtp:
try:
@ -138,18 +139,18 @@ class EmailMFA(mfas.MFA):
msg['From'] = self.fromEmail.cleanStr()
msg['To'] = identifier
msg.attach(MIMEText(f'Your verification code is {code}', 'plain'))
msg.attach(MIMEText(f'A login attemt has been made from {request.ip}.\nTo continue, provide the verification code {code}', 'plain'))
if self.enableHTML.value:
msg.attach(MIMEText(f'<p>Your OTP code is <b>{code}</b></p>', 'html'))
msg.attach(MIMEText(f'<p>A login attemt has been made from <b>{request.ip}</b>.</p><p>To continue, provide the verification code <b>{code}</b></p>', 'html'))
smtp.sendmail(self.fromEmail.value, identifier, msg.as_string())
except smtplib.SMTPException as e:
logger.error('Error sending email: {}'.format(e))
raise
def sendCode(self, userId: str, username: str, identifier: str, code: str) -> mfas.MFA.RESULT:
self.doSendCode(identifier, code)
def sendCode(self, request: 'ExtendedHttpRequest', userId: str, username: str, identifier: str, code: str) -> mfas.MFA.RESULT:
self.doSendCode(request, identifier, code,)
return mfas.MFA.RESULT.OK
def login(self) -> smtplib.SMTP:

7
server/src/uds/mfas/Sample/mfa.py
View File

@ -9,6 +9,7 @@ from uds.core.ui import gui
if typing.TYPE_CHECKING:
from uds.core.module import Module
from uds.core.util.request import ExtendedHttpRequest
logger = logging.getLogger(__name__)
@ -33,8 +34,8 @@ class SampleMFA(mfas.MFA):
def label(self) -> str:
return 'Code is in log'
def sendCode(self, userId: str, username: str, identifier: str, code: str) -> mfas.MFA.RESULT:
logger.debug('Sending code: %s', code)
def sendCode(self, request: 'ExtendedHttpRequest', userId: str, username: str, identifier: str, code: str) -> mfas.MFA.RESULT:
logger.debug('Sending code: %s (from %s)', code, request.ip)
return mfas.MFA.RESULT.OK

24
server/src/uds/web/views/modern.py
View File

@ -208,16 +208,26 @@ def mfa(request: ExtendedHttpRequest) -> HttpResponse:
request.authorized = True
return HttpResponseRedirect(reverse('page.index'))
# Not allowed to login, redirect to login error page
logger.warning('MFA identifier not found for user %s on authenticator %s. It is required by MFA %s', request.user.name, request.user.manager.name, mfaProvider.name)
logger.warning(
'MFA identifier not found for user %s on authenticator %s. It is required by MFA %s',
request.user.name,
request.user.manager.name,
mfaProvider.name,
)
return errors.errorView(request, errors.ACCESS_DENIED)
if request.method == 'POST': # User has provided MFA code
form = MFAForm(request.POST)
if form.is_valid():
code = form.cleaned_data['code']
try:
mfaInstance.validate(
userHashValue, request.user.name, mfaIdentifier, code, validity=validity
request,
userHashValue,
request.user.name,
mfaIdentifier,
code,
validity=validity,
)
request.authorized = True
# Remove mfa_start_time from session
@ -245,7 +255,13 @@ def mfa(request: ExtendedHttpRequest) -> HttpResponse:
else:
# Make MFA send a code
try:
result = mfaInstance.process(userHashValue, request.user.name, mfaIdentifier, validity=validity)
result = mfaInstance.process(
request,
userHashValue,
request.user.name,
mfaIdentifier,
validity=validity,
)
if result == mfas.MFA.RESULT.ALLOWED:
# MFA not needed, redirect to index after authorization of the user
request.authorized = True