Source code for fleetingform.views
import base64
import logging
import datetime
from decimal import Decimal
from django.core.exceptions import ObjectDoesNotExist
from django.views import View
from django.views.generic.base import RedirectView
from django.template.response import TemplateResponse
from django.shortcuts import get_object_or_404, redirect
from django.contrib import messages
from django.conf import settings
from rest_framework import generics, mixins
from rest_framework.authentication import TokenAuthentication
from fleetingform.models import FleetingForm, FleetingNamespace
from fleetingform.forms import (UserAuthenticationForm,
PasswordAuthenticationForm,
UserPasswordAuthenticationForm, )
from fleetingform.serializers import (FleetingFormSerializer,
FleetingNamespaceSerializer, )
from fleetingform.authentication import FleetingAuthentication
from fleetingform.permissions import (FleetingFormTokenPermission,
FleetingNamespaceTokenPermission,
FleetingFormHardLimitPermission, )
from fleetingform.lib import auth_token_field, auth_username_field
from fleetingform.lib.form_generator import generate_fleeting_form_class_for
from fleetingform.errors import (FleetingFormGenerationError,
FleetingOTPRetriesExceeded,
FleetingAuthOTPError, )
logger = logging.getLogger(__name__)
[docs]def send_otp_and_set_messages(user, fform, request):
try:
user.generate_and_send_otp()
except FleetingOTPRetriesExceeded:
logger.error('{} OTP retries exceeded for {}'.format(
fform.code,
user.username))
messages.error(request,
"Too many OTP retries for this form, "
"code not sent.")
except FleetingAuthOTPError:
messages.warning(request,
"Failed to send the OTP. "
"Please try again in a few minutes.")
except Exception:
messages.error(request,
"Error while sending OTP. "
"Please try again in a few minutes.")
else:
return True
return False
[docs]class FleetingFormListCreateView(generics.ListCreateAPIView):
serializer_class = FleetingFormSerializer
authentication_classes = [FleetingAuthentication, ]
permission_classes = [FleetingFormTokenPermission,
FleetingFormHardLimitPermission, ]
[docs] def get_queryset(self):
qs = FleetingForm.objects.none()
if self.request.namespace: # pragma: no branch - auth fail up stack
qs = self.request.namespace.forms.all()
return qs
[docs]class FleetingFormRetrieveDestroyView(generics.RetrieveDestroyAPIView):
serializer_class = FleetingFormSerializer
permission_classes = [FleetingFormTokenPermission, ]
authentication_classes = [FleetingAuthentication, ]
[docs] def get_queryset(self):
qs = FleetingForm.objects.none()
if self.request.namespace: # pragma: no branch - auth fail up stack
qs = self.request.namespace.forms.all()
return qs
[docs]class FleetingNamespaceListCreateView(generics.ListCreateAPIView):
serializer_class = FleetingNamespaceSerializer
permission_classes = [FleetingNamespaceTokenPermission, ]
authentication_classes = [TokenAuthentication, ]
[docs] def get_queryset(self):
qs = FleetingNamespace.objects.none()
if self.request.namespace:
qs = FleetingNamespace.objects.filter(id=self.request.namespace.id)
elif (self.request.user and # pragma: no branch - authd
self.request.user.is_authenticated):
qs = self.request.user.namespaces.all()
return qs
[docs]class FleetingNamespaceRetrieveView(generics.RetrieveUpdateAPIView): # pragma: no cover - deprecated
serializer_class = FleetingNamespaceSerializer
permission_classes = [FleetingNamespaceTokenPermission, ]
authentication_classes = [FleetingAuthentication, TokenAuthentication]
[docs] def get_queryset(self):
qs = FleetingNamespace.objects.none()
if self.request.namespace:
qs = FleetingNamespace.objects.filter(id=self.request.namespace.id)
elif (self.request.user and # pragma: no branch - authd
self.request.user.is_authenticated):
qs = self.request.user.namespaces.all()
return qs
[docs]class FleetingNamespaceRetrieveUpdateView(mixins.RetrieveModelMixin,
mixins.UpdateModelMixin,
generics.GenericAPIView):
serializer_class = FleetingNamespaceSerializer
permission_classes = [FleetingNamespaceTokenPermission, ]
authentication_classes = [FleetingAuthentication, TokenAuthentication]
[docs] def get_queryset(self):
qs = FleetingNamespace.objects.none()
if self.request.namespace:
qs = FleetingNamespace.objects.filter(id=self.request.namespace.id)
elif (self.request.user and # pragma: no branch - authd
self.request.user.is_authenticated):
qs = self.request.user.namespaces.all()
return qs
[docs] def put(self, request, *args, **kwargs): # pragma: no cover - builtin
return self.update(request, *args, **kwargs)
[docs]class FleetingFormLoginView(View):
"""Handle Fleeting Form Logins
Overall workflow depends on the authentication type configured.
1. Username only:
- Display username form.
- on match proceed
- on fail return to
2. Password only:
- Display password form
- on match proceed
- on fail return to
3. Username and Password:
1. Username and password provided:
- Display Userpass form
- on match proceed
- on fail return to
1. Username and otp:
- Display Username form
- on match, continue
- Display password form
- on match, proceed
- on cancel, back to username form
- on fail return to
Workflows
USERNAME
- sets auth_token
PASSWORD
- sets auth_token
USERPASS
- sets auth_token
USERPASS_OTP
- sets auth_username
- sets auth_token
"""
username_instructions = "Unlock this form with a username."
userpass_instructions = "Unlock this form with a username and password."
password_instructions = "Unlock this form with a password."
otp_instructions = "Unlock this form with your one time code."
def _authenticated(self, fform, request):
"""Check if this request has been authenticated for this form.
:param fform: the form to check authentication
:type fform: FleetingForm
:param request: the request to check authentication
:type request: django.http.HttpRequest
:returns: True if previously authenticated, else False
:rtype: bool
"""
token = fform.auth_token
if token:
if str(token) == request.session.get(auth_token_field(fform)):
return True
return False
def _get_current_user(self, fform, request):
if fform.auth.requires_username:
username = request.session.get(auth_username_field(fform))
if username:
return fform.auth.users.get(username=username)
return fform.auth.users.first()
def _get_template_vars(self, fform, request):
"""Get the authentication form template variables.
:param fform: the form to check authentication
:type fform: FleetingForm
:returns: template variables
:rtype: dict
"""
_vars = {
'title': fform.auth.title,
'content': fform.auth.content,
'action': fform.auth.action
}
if fform.auth.otp:
user = self._get_current_user(fform, request)
remaining = settings.FLEETING_OTP_MAX_ATTEMPTS + 1 - user.attempts
otp_vars = {
'otp': True,
'remaining_codes': remaining,
'fform_code': fform.code,
'otp_destination': user.otp_contact_obscured
}
_vars = {**_vars, **otp_vars}
return _vars
def _get_form(self, fform, request):
"""Get the right form for the current auth type and populate.
Before entering this method be sure to verify that the auth
type isnt none - why would you be here if it was?
:param fform: the fleeting form
:type fform: FleetingForm
:param request: the request
:type request: django.http.HttpRequest
:returns: authentication form for current stage
:rtype: django.forms.Form
"""
form, instructions = None, None
post = request.POST or None
username = request.session.get(auth_username_field(fform), '')
try:
user = fform.auth.users.get(username=username)
except ObjectDoesNotExist:
user = None
# If this is a two stage request, handle it.
if fform.auth.username_and_otp:
if user: # The user is set in the session
if not user.password: # An OTP has never been sent
send_otp_and_set_messages(user, fform, request)
post = post if post and 'password' in post else None
form = PasswordAuthenticationForm(fform, user.username, post)
instructions = self.otp_instructions
else:
form = UserAuthenticationForm(fform, post)
instructions = self.username_instructions
elif fform.auth.otp:
user = user or fform.auth.users.first()
if not user.password:
send_otp_and_set_messages(user, fform, request)
form = PasswordAuthenticationForm(fform, user.username, post)
instructions = self.otp_instructions
elif fform.auth.username_and_password:
# Straight up username and password
form = UserPasswordAuthenticationForm(fform, post)
instructions = self.userpass_instructions
elif fform.auth.password_only:
# Password only.
form = PasswordAuthenticationForm(fform, username, post)
instructions = self.password_instructions
elif fform.auth.username_only:
# Username only authentication
form = UserAuthenticationForm(fform, post)
instructions = self.username_instructions
else:
logger.error("Failed to find form for {} ({}).".format(
fform, fform.auth.type))
raise FleetingFormGenerationError("Unknown form type.")
return (form, instructions)
[docs] def get(self, request, code):
"""Handle a get request for a Fleeting Form Login.
Checks is the form requires authentication. If so, and not
authenticated, renders login. If not, or authenticated,
redirects to form.
:param request: the request
:type request: django.http.HttpRequest
:param code: the one time code for the form extracted from the url
:type code: str
:returns: If auth required redender, else redirect to form
:rtype: django.views.generic.base.RedirectView or
django.template.response.TemplateResponse
"""
fform = get_object_or_404(FleetingForm, code=code)
if not fform.auth.required or self._authenticated(fform, request):
return redirect('form-display', code=code)
# This can raise but never should because the auth type has been
# validated. If it does, let it get to the user.
form, instructions = self._get_form(fform, request)
logger.info(f"{fform.code}: get request for /login {fform.auth.type}")
return TemplateResponse(request,
'fleetingform/login.html',
{'form': form,
**self._get_template_vars(fform, request)})
[docs] def post(self, request, code):
"""Handle a get request for a Fleeting Form Login.
Check if the authentication form is valid. If so, set session
variables and either redirect to the form or render the next
authentication stage.
:param request: the request
:type request: django.http.HttpRequest
:param code: the one time code for the form extracted from the url
:type code: str
:returns: If further auth required or invalid render,
else redirect to form
:rtype: django.views.generic.base.RedirectView or
django.template.response.TemplateResponse
"""
fform = get_object_or_404(FleetingForm, code=code)
form, instructions = self._get_form(fform, request)
logger.info(f"{fform.code}: post request for /login {fform.auth.type}")
if form.is_valid():
if fform.auth.username_and_otp:
# Stage1: Completing username field
if form.fields.get('username'):
request.session[auth_username_field(fform)] = \
form.cleaned_data['username']
# Username set in the session, the password form returned.
form, instructions = self._get_form(fform, request)
return TemplateResponse(request,
'fleetingform/login.html',
{'form': form,
**self._get_template_vars(
fform, request)})
# Stage2: Completing the password field
request.session[auth_token_field(fform)] = str(fform.auth_token)
logger.debug(f"{fform.code}: login complete {fform.auth.type}")
return redirect('form-display', code=code)
return TemplateResponse(request,
'fleetingform/login.html',
{'form': form,
**self._get_template_vars(fform, request)})
[docs]class UserFormView(View):
"""User facing forms.
This view is responsible for rendering the forms and handling
user input.
"""
[docs] def decode_query_params(self, params={}):
"""Decode query params.
Query params can be sent as base64 encoded values when prefixed with
``b64:``. Decode query params into a flat dict of decoded
``key: value`` pairs.
:param params: GET and/or POST querydict contents
:type params: dict-like
:returns: all decoded key: value pairs.
:rtype: dict
"""
initial_data = {}
for key, value in params.items():
initial_data[key] = value
if value.startswith("b64:"):
encoding, encoded = value.split(":", 1)
logger.debug(f'encoding: encoded - {encoding}:{encoded}')
try:
decoded = base64.urlsafe_b64decode(encoded)
initial_data[key] = decoded.decode('utf-8')
logger.debug(f'initial_data[{key}] = {decoded} {encoding}:{encoded}')
except Exception as e:
logger.error(f"failed to decode '{value}': {e}.")
del initial_data[key]
return initial_data
[docs] def get(self, request, code):
"""Handle a get request for a Fleeting Form.
Checks is the form requires authentication. If so, and not
authenticated, redirect to login workflow. If not, or authenticated,
render dynamically generated form.
:param request: the request
:type request: django.http.HttpRequest
:param code: the one time code for the form extracted from the url
:type code: str
:returns: If auth required redirect, else render form
:rtype: django.views.generic.base.RedirectView or
django.template.response.TemplateResponse
"""
fform = get_object_or_404(FleetingForm, code=code)
if fform.auth and fform.auth.required:
session_auth_token = request.session.get(
auth_token_field(fform))
if session_auth_token != str(fform.auth_token):
return redirect('form-login', code=code)
logger.info(f"{fform.code}: post request [{fform.template.type}]")
try:
initial_data = {}
if fform.completed:
messages.success(request, "This form has been completed.")
initial_data = fform.result
else:
initial_data = self.decode_query_params(request.GET)
form_class = generate_fleeting_form_class_for(
fform.template.form_controls.all(),
initial_data)
form = form_class()
template = {
'form': form,
'actions': fform.template.actions.values_list(
'label', flat=True),
'title': fform.template.title,
'content': fform.template.content,
'completed': fform.completed,
'result_action': fform.result.get('action', ''),
'namespace': fform.namespace}
fform.open()
except Exception as e:
# Notify someone
fform.error('pre-render-failed', str(e))
raise
return TemplateResponse(
request, fform.template.html_template, template)
[docs] def post(self, request, code):
"""Handle a post request for a Fleeting Form.
Checks is the form requires authentication. If so, and not
authenticated, redirect to login workflow. If not, or authenticated,
create dynamically generated form and check validity.
If valid, save and complete form. If invalid, render form with errors.
:param request: the request
:type request: django.http.HttpRequest
:param code: the one time code for the form extracted from the url
:type code: str
:returns: If auth required redirect, else render form.
:rtype: django.views.generic.base.RedirectView or
django.template.response.TemplateResponse
"""
fform = get_object_or_404(FleetingForm, code=code)
if fform.auth and fform.auth.required:
if request.session.get(auth_token_field(fform)) != \
str(fform.auth_token):
return redirect('form-login', code=code)
logger.info(f"{fform.code}: post request [{fform.template.type}]")
try:
form_class = generate_fleeting_form_class_for(
fform.template.form_controls.all(),
self.decode_query_params(request.GET))
logger.debug(f"{fform.code}: new POST {request.POST}")
form = form_class(request.POST)
if not fform.completed and form.is_valid():
logger.debug(
f"{fform.code}: fform not complete and valid, saving.")
result = form.cleaned_data
result['action'] = request.POST.get('action', 'unknown')
fform.complete(self._jsonify_result(result))
else:
logger.debug(f"{fform.code}: complete or invalid.")
template = {
'form': form,
'actions': fform.template.actions.values_list('label',
flat=True),
'title': fform.template.title,
'content': fform.template.content,
'completed': fform.completed,
'result_action': fform.result.get('action', ''),
'namespace': fform.namespace, }
if fform.completed:
messages.success(request, "This form has been completed.")
except Exception as e:
fform.error('post-failed', str(e))
raise
return TemplateResponse(
request, fform.template.html_template, template)
def _jsonify_result(self, result):
"""Because serializing dates and times isn't a standard thing..."""
out = {}
for key, value in result.items():
if isinstance(value, datetime.datetime):
out[key] = value.isoformat(timespec='seconds')
elif isinstance(value, datetime.date):
out[key] = value.isoformat()
elif isinstance(value, datetime.time):
out[key] = value.isoformat()
elif isinstance(value, Decimal):
out[key] = str(value)
elif isinstance(value, dict):
out[key] = self._jsonify_result(value)
else:
out[key] = value
return out
[docs]class FleetingOTPResetRedirectView(RedirectView):
"""Reset and resend the OTP for a form.
For two stage auth types, silently noops for sessions in which
the username hasn't been verified yet.
Otherwise, forces a new passcode to be generated and sent.
"""
permanent = False
pattern_name = 'form-login'
def _get_user_from_session(self, fform):
username = self.request.session.get(
auth_username_field(fform))
# If there is a username in the session and it matches, resend
if username:
return fform.auth.users.get(username=username)
return None
[docs] def get_redirect_url(self, *args, **kwargs):
"""If this is an OTP secured form, and the user is set if required,
reset and resend the otp, then redirect to the form login view."""
fform = get_object_or_404(FleetingForm, code=kwargs['code'])
logger.debug(f"{fform.code}: new otp requested.")
# Does this form use some kind of OTP?
if fform.auth.email_otp or fform.auth.phone_otp:
# Does that OTP type require a username as well?
if fform.auth.username_and_otp:
user = self._get_user_from_session(fform)
if user: # pragma: no branch
send_otp_and_set_messages(user, fform, self.request)
# Otherwise, user isn't set, noop and send the user back to
# the begging (username form) of the auth workflow.
else:
logger.warn(f"{fform.code}: resend requested but no user.")
else:
# No username required, resend to the lone user.
user = fform.auth.users.first()
send_otp_and_set_messages(user, fform, self.request)
return super().get_redirect_url(*args, **kwargs)