import uuid
import random
import logging
from datetime import timedelta
from django.db import models
from django.utils import timezone
from django.utils.module_loading import import_string
from django.utils.crypto import get_random_string
from django.conf import settings
from django.core.validators import (URLValidator,
MaxValueValidator,
MinValueValidator,
RegexValidator, )
from django.core.exceptions import ObjectDoesNotExist, ValidationError
from django.contrib.postgres.fields import JSONField
from django.contrib.contenttypes.models import ContentType
from django.contrib.contenttypes.fields import (GenericForeignKey,
GenericRelation)
from phonenumber_field.modelfields import PhoneNumberField
from fleetingform.errors import (FleetingFormCompleteError,
FleetingAuthOTPError,
FleetingOTPRetriesExceeded,
FleetingDeletionError, )
from fleetingform.lib import (send_otp_sms,
send_otp_email)
logger = logging.getLogger(__name__)
def _next_code():
"""Get the next random one time code for a FleetingForm.
:returns: code of length ``settings.FLEETING_CODE_LENGTH``
"""
code = get_random_string(length=settings.FLEETING_CODE_LENGTH)
while FleetingForm.objects.filter(code=code):
code = get_random_string(length=settings.FLEETING_CODE_LENGTH)
return code
[docs]class FleetingNamespace(models.Model):
"""Namespaces are a honking great idea - group forms and settings.
Each namespace groups FleetingForms that require consistent settings
ans styling, i.e. those supporting a single application or user.
All forms in a namespace share the same subdomain
(https://subdomain.fleetingforms.io/), styling, support email.
Users access the a namespace by providing the correct token in the
``settings.FLEETING_TOKEN_HEADER`` HTTP header.
"""
RE_URL_UNSAFE = r'[\[\]<>#%";/?:@&=+$,{}|.`\'^\\ \t\r\n\f]'
CH_URL_UNSAFE = (r'< > # % " ; / ? : @ & = + $ , { } | \ ^ [ ] ` \' .')
URL_FLEETI_NG = "fleeti.ng"
URL_FLEETING_LINK = "fleeting.link"
URL_SHORTENERS = (
(URL_FLEETI_NG, URL_FLEETI_NG),
(URL_FLEETING_LINK, URL_FLEETING_LINK),
)
user = models.ForeignKey(settings.AUTH_USER_MODEL,
on_delete=models.CASCADE,
related_name='namespaces')
token = models.UUIDField(default=uuid.uuid4, )
subdomain = models.CharField(
max_length=64,
unique=True,
validators=[RegexValidator(
regex=RE_URL_UNSAFE,
message="Only URL safe characers are allowed. "
"Don't include whitespace or any of "
"'{}'.".format(CH_URL_UNSAFE),
inverse_match=True)])
url_shortener = models.CharField(max_length=16,
choices=URL_SHORTENERS,
default=URL_FLEETI_NG)
style = models.URLField(blank=True, null=True,
validators=[URLValidator(
schemes=['http', 'https']
)])
logo = models.URLField(blank=True, null=True,
validators=[URLValidator(
schemes=['http', 'https']
)])
retention = models.PositiveIntegerField(
default=14,
validators=[
MaxValueValidator(settings.FLEETING_MAX_RETENTION),
MinValueValidator(1)])
support_email = models.EmailField()
soft_limit = models.PositiveIntegerField(default=100)
hard_limit = models.PositiveIntegerField(default=100)
# webhooks = ManyToThisField(FleetingWebhook)
[docs] @classmethod
def from_request(cls, request):
"""Get the namespace from a Django Request object.
:param request: the request to extract the namespace from.
:type request: Django Request
:returns: FleetingNamespace that matches the token or ``None``
:rtype: FleetingNamespace or None
"""
token = request.META.get(settings.FLEETING_TOKEN_HEADER)
if token:
try:
namespace = cls.objects.get(token=token)
return namespace
except ObjectDoesNotExist:
logger.warn(f"namespace doest exist for {token}")
except (ValueError, ValidationError):
logger.warn(f"Invalid UUID format for {token}")
return None
@property
def total_forms(self):
return self.audit_entries.count()
@property
def active_forms(self):
return self.forms.count()
@property
def forms_this_month(self):
return self.total_forms_since(timezone.now().replace(
day=1, hour=0, minute=0, second=0, microsecond=0))
@property
def forms_last_month(self):
end_last_billing = (timezone.now().replace(
day=1, hour=23, minute=59, second=59,
microsecond=999999) -
timedelta(days=1))
start_last_billing = (end_last_billing.replace(
day=1, hour=0, minute=0, second=0,
microsecond=0) -
timedelta(days=1))
return self.total_forms_between(start_last_billing, end_last_billing)
[docs] def save(self, *args, **kwargs):
self.full_clean()
return super().save(*args, **kwargs)
[docs] def delete(self, *args, **kwargs):
raise FleetingDeletionError(self.__class__.__name__, self.pk)
def __str__(self):
return "{}.{} [{}]".format(self.subdomain,
settings.FLEETING_DOMAIN,
self.id)
[docs]class FleetingWebhook(models.Model):
"""Webhooks for namespaces.
Enables a user-configurable callback when a form changes state.
"""
WEBHOOK_EVENT_CREATE = 'create'
WEBHOOK_EVENT_COMPLETE = 'complete'
WEBHOOK_EVENT_ERROR = 'error'
WEBHOOK_EVENT_ACCESS = 'access'
WEBHOOK_EVENTS = (
(WEBHOOK_EVENT_CREATE, "Created"),
(WEBHOOK_EVENT_COMPLETE, "Completed"),
(WEBHOOK_EVENT_ERROR, "Error"),
(WEBHOOK_EVENT_ACCESS, "On Access")
)
name = models.CharField(max_length=128, null=True, blank=True, default='')
event = models.CharField(max_length=8,
choices=WEBHOOK_EVENTS,
default=WEBHOOK_EVENT_CREATE)
namespace = models.ForeignKey(FleetingNamespace,
on_delete=models.CASCADE,
related_name="webhooks")
url = models.URLField(
validators=[URLValidator(
schemes=['https'],
message="Webhooks must have https:// urls."), ])
token = models.CharField(max_length=256, blank=True, default='')
def __str__(self):
return "({}): {} - {} to {} [{}]".format(
self.namespace, self.name, self.event, self.url, self.id)
[docs]class FleetingTemplate(models.Model):
"""Templates control how the form is displayed and which actions are taken.
Every form has a template. The template defines which HTML page will
be rendered and which actions will be taken when the user submits the
form.
The simplest template, 'generic', renders a title, introductory content,
and a single form. On submit, the contents of the form are saved in the
results.
"""
TEMPLATE_TYPE_GENERIC = 'generic'
TEMPLATE_TYPE_STRIPE = 'stripe'
TEMPLATE_TYPE_PAYPAL = 'paypal'
TEMPLATE_TYPES = (
(TEMPLATE_TYPE_GENERIC, 'Generic'),
(TEMPLATE_TYPE_STRIPE, 'Stripe'),
(TEMPLATE_TYPE_PAYPAL, 'Paypal'),
)
TEMPLATE_HELPERS = {
TEMPLATE_TYPE_GENERIC:
'fleetingform.template_helpers.generic.GenericTemplateHelper',
}
TEMPLATE_CONTENT_TYPE_MARKDOWN = 'md'
TEMPLATE_CONTENT_TYPE_RESTRUCTURED = 'rst'
TEMPLATE_CONTENT_TYPE_PLAINTEXT = 'txt'
TEMPLATE_CONTENT_TYPES = (
(TEMPLATE_CONTENT_TYPE_MARKDOWN, "Markdown"),
(TEMPLATE_CONTENT_TYPE_RESTRUCTURED, "ReStructured Text"),
(TEMPLATE_CONTENT_TYPE_PLAINTEXT, "Plain Text")
)
type = models.CharField(max_length=16, choices=TEMPLATE_TYPES)
title = models.CharField(max_length=200, blank=False)
content = models.CharField(max_length=5000, null=True,
blank=True, default='')
content_type = models.CharField(max_length=3,
choices=TEMPLATE_CONTENT_TYPES,
default=TEMPLATE_CONTENT_TYPE_PLAINTEXT)
params = JSONField(default=dict)
form_controls = GenericRelation('FleetingFormControl')
@property
def helper(self):
"""Get the helper class name for this template.
:returns: helper class full dotted path.
:rtype: str
"""
return self.TEMPLATE_HELPERS.get(self.type)
@property
def supported_templates(self):
"""Get the list of supported templates.
:returns: supported template names.
:rtype: [str, ]
"""
return [t[0] for t in self.TEMPLATE_TYPES]
@property
def html_template(self):
"""Get the HTML template path for this Template
:returns: full static HTML template path.
:rtype: str
"""
return "fleetingform/{}.html".format(self.type)
[docs] def helper_class(self):
"""Get the helper class.
:returns: class object for this template type's helper.
:rtype: class
:raises: ImportError if helper cannot be found.
"""
return import_string(self.helper)
[docs] @classmethod
def template_helper_class_for(cls, _type):
"""Get the helper class for any supported template type.
:returns: class object for this template type's helper.
:rtype: class
:raises: KeyError if the type isn't valid,
ImportError if helper cannot be found.
"""
return import_string(cls.TEMPLATE_HELPERS[_type])
def __str__(self):
return "{} ({}) [{}]".format(self.title, self.type, self.id)
[docs]class FleetingAuth(models.Model):
"""Auth controls the authentication workflow for the form.
Every form has authentication parameters. Authentication parameters
define how the authentication pages are displayed and which authentication
workflow the user has to complete to access the form.
The simplest form of authentication, ``none``, does not require any user
authentication.
The most complex form, ``user_otp_phone``, supports multiple users with
individual one time pass codes delivered over SMS.
"""
AUTH_DEFAULT_TITLE = "Unlock Fleeting Form"
AUTH_DEFAULT_CONTENT = "Use your credentials to unlock the form."
AUTH_TYPE_NONE = 'none'
AUTH_TYPE_USER = 'user'
AUTH_TYPE_PASS = 'pass'
AUTH_TYPE_USER_PASS = 'user_pass'
AUTH_TYPE_PASS_OTP_EMAIL = 'otp_email'
AUTH_TYPE_PASS_OTP_PHONE = 'otp_phone'
AUTH_TYPE_USER_PASS_OTP_EMAIL = 'user_otp_email'
AUTH_TYPE_USER_PASS_OTP_PHONE = 'user_otp_phone'
AUTH_TYPES = (
(AUTH_TYPE_NONE, "None"),
(AUTH_TYPE_USER, "Username"),
(AUTH_TYPE_PASS, "Password"),
(AUTH_TYPE_USER_PASS, "Username and Password"),
(AUTH_TYPE_PASS_OTP_EMAIL, "One-time Passcode via Email"),
(AUTH_TYPE_PASS_OTP_PHONE, "One-time Passcode via SMS"),
(AUTH_TYPE_USER_PASS_OTP_EMAIL,
"Username and One-time Passcode via Email"),
(AUTH_TYPE_USER_PASS_OTP_PHONE,
"Username and One-time Passcode via SMS"),
)
AUTH_TYPES_USERNAME = (AUTH_TYPE_USER, )
AUTH_TYPES_PASSWORD = (AUTH_TYPE_PASS, )
AUTH_TYPES_USERPASS = (AUTH_TYPE_USER_PASS, )
AUTH_TYPES_USEROTP = (AUTH_TYPE_USER_PASS_OTP_EMAIL,
AUTH_TYPE_USER_PASS_OTP_PHONE, )
AUTH_TYPES_EMAIL_OTP = (AUTH_TYPE_PASS_OTP_EMAIL,
AUTH_TYPE_USER_PASS_OTP_EMAIL)
AUTH_TYPES_PHONE_OTP = (AUTH_TYPE_PASS_OTP_PHONE,
AUTH_TYPE_USER_PASS_OTP_PHONE)
AUTH_TYPES_OTP = AUTH_TYPES_PHONE_OTP + AUTH_TYPES_EMAIL_OTP
AUTH_TYPES_OTP_ONLY = (AUTH_TYPE_PASS_OTP_EMAIL,
AUTH_TYPE_PASS_OTP_PHONE, )
AUTH_TYPES_USER = (AUTH_TYPES_USERNAME + AUTH_TYPES_USERPASS +
AUTH_TYPES_USEROTP)
type = models.CharField(max_length=16,
choices=AUTH_TYPES,
default=AUTH_TYPE_NONE,)
title = models.CharField(max_length=120, blank=True,
default='')
content = models.CharField(max_length=1000, blank=True,
default='')
# users = models.ManyToThisField(FleetingUser)
action = models.CharField(max_length=30, default='Unlock')
form_controls = GenericRelation('FleetingFormControl')
@property
def required(self):
"""Does this form require authentication?
:returns: True if authentication required, else False
:rtype: bool
"""
return self.type != self.AUTH_TYPE_NONE
# Type Helpers - These functions are used to get at the general
# auth type.
@property
def username_only(self):
"""Does this form require a username only.
:returns: True if username only authentication required, else False
:rtype: bool
"""
return self.type in self.AUTH_TYPES_USERNAME
@property
def password_only(self):
"""Does this form require a password only.
:returns: True if password only authentication required, else False
:rtype: bool
"""
return self.type in self.AUTH_TYPES_PASSWORD
@property
def username_and_password(self):
"""Does this form require a username and static password.
:returns: True if username and static password required, else False
:rtype: bool
"""
return self.type in self.AUTH_TYPES_USERPASS
@property
def username_and_otp(self):
"""Does this form require a username and one time passcode?
:returns: True if username and static password required, else False
:rtype: bool
"""
return self.type in self.AUTH_TYPES_USEROTP
@property
def email_otp(self):
"""Does this form use email for the one time pass?
:returns: True if username and email one time pass required, else False
:rtype: bool
"""
return self.type in self.AUTH_TYPES_EMAIL_OTP
@property
def phone_otp(self):
"""Does this form use phone for the one time pass?
:returns: True if username and phone one time pass required, else False
:rtype: bool
"""
return self.type in self.AUTH_TYPES_PHONE_OTP
@property
def otp(self):
"""Does this form require a one time pass?
:returns: True if one time pass required, else False
:rtype: bool
"""
return self.type in self.AUTH_TYPES_OTP
@property
def requires_password(self):
"""Does this form require a password of some kind? OTP or Static.
:returns: True if password required, else False
:rtype: bool
"""
return self.type not in (self.AUTH_TYPE_NONE, self.AUTH_TYPE_USER, )
@property
def requires_username(self):
"""Does this form require a username?
:returns: True if username required, else False
:rtype: bool
"""
return self.type in self.AUTH_TYPES_USER
[docs] def authenticate(self, username="", password=""):
"""Authenticate a user and password.
Given the auth type, authenticate the given username and password.
:param username: the username to authenticate.
:type username: str
:param password: the plain text password to authenticate.
:type password: str
:returns: True if username and password valid, else False
:rtype: bool
"""
if self.type == self.AUTH_TYPE_NONE:
return True
if not self.users.count():
return False
if self.username_only:
if not username:
return False
return self.verify_username(username=username)
if self.password_only:
if not password:
return False
return self.verify_password(password, self.users.first().password)
# If it isn't none, user, or pass, it must be username and password.
if not (username and password):
return False
try:
user = self.users.get(username=username)
except ObjectDoesNotExist:
return False
return self.verify_password(password, user.password)
[docs] def verify_username(self, username=""):
"""Verify that a username is valid for the form.
Checks whether the username is present in ``users``.
:param username: username to verify
:type username: str
:returns: True if user is present, else False
:rtype: bool
"""
return bool(self.users.filter(username=username))
[docs] def verify_password(self, password, password_hash):
"""Verify that a plain text password matches a hash.
Automatically identifies and instantiates the correct hasher
and performs the hash verification.
:param password: password to verify
:type password: str
:returns: True if password matches hash, else False
:rtype: bool
"""
try:
_, algo, it, salt, hsh = password_hash.split("$")
except (IndexError, ValueError) as e:
logger.error("Bad password hash in verify_password: {}".format(
password_hash))
raise ValueError("Password hash format not supported.") from e
hasher = settings.FLEETING_HASHERS.get(algo)
return hasher.verify(password, password_hash)
[docs] def opened_by(self, username):
"""Set the opened_by timestamp for a user on access.
:param username: the user that opened the form.
:type username: str
:returns: None
"""
try:
user = self.users.get(username=username)
except ObjectDoesNotExist:
return
if not user.opened_on:
user.opened_on = timezone.now()
user.save()
def __str__(self):
return "{} [{}]".format(
self.get_type_display(),
self.id, )
[docs]class FleetingUser(models.Model):
class Meta:
unique_together = ('auth', 'username', )
ordering = ('id', )
auth = models.ForeignKey(FleetingAuth,
on_delete=models.CASCADE,
related_name='users')
username = models.CharField(max_length=60, blank=True, default='')
password = models.CharField(max_length=130, blank=True, default='')
email = models.EmailField(blank=True, default='')
phone = PhoneNumberField(blank=True, default='')
opened_on = models.DateTimeField(null=True, default=None)
attempts = models.PositiveIntegerField(default=0)
@property
def otp_contact_obscured(self):
"""The contact information used to send OTP, obscured.
Hides the email or phone number with some x's.
:returns: email or phone for this user with some x's
:rtype: str
"""
if self.auth.email_otp:
name, domain = self.email.split("@")
domain, tld = domain.rsplit(".", 1)
name_hidden = int(len(name) / 2)
name_shown = len(name) - name_hidden
name = name[:name_shown] + ("x" * name_hidden)
domain_hidden = int(len(domain) / 2)
domain_shown = len(domain) - domain_hidden
domain = ("x" * domain_hidden) + domain[domain_shown-1:]
email = f"{name}@{domain}.{tld}"
return email
elif self.auth.phone_otp:
strphone = str(self.phone)
length = len(str(strphone))
hide_amount = int(length / 3)
left_amount = int((length - hide_amount) / 2)
right_amount = length - (hide_amount + left_amount)
x = "x" * hide_amount
return "{}{}{}".format(
strphone[:left_amount], x, strphone[-right_amount:])
return ''
[docs] def generate_and_send_otp(self):
"""Generate and send a new one time passcode to the user.
:raises: FleetingAuthOTPError on OTP send failure.
"""
if self.attempts > settings.FLEETING_OTP_MAX_ATTEMPTS:
logger.error(f"{self.auth.form.code}: otp resends exceeded for "
f"{self.username} [{self.attempts}]")
raise FleetingOTPRetriesExceeded(self)
otp = "{:06}".format(random.randint(10000, 999999))
if self.auth.email_otp:
send_otp_email(self.email, otp, self.auth.form)
elif self.auth.phone_otp:
send_otp_sms(str(self.phone), otp, self.auth.form)
else:
logger.error(f"{self.auth.form.code}: otp send failed for "
f"{self.username}")
raise FleetingAuthOTPError(self, self.auth.type)
self.encrypt_and_set_password(otp)
self.attempts += 1
self.save()
[docs] def encrypt_and_set_password(self, plain_password):
"""Encrypt and set the user password.
:param plain_password: the password to encrypt and set.
:type plain_password: str
"""
self.password = settings.FLEETING_DEFAULT_HASHER.hash(plain_password)
[docs] def authenticate(self, plain_password):
"""Authenticate this user with plain_password.
:param plain_password: the plain text password to authenticate
:type plain_password: str
:returns: True is the password matches, else False
:rtype: bool
"""
return self.auth.authenticate(
username=self.username, password=plain_password)
def __str__(self):
return "{}".format(self.username)
[docs]class FleetingAction(models.Model):
"""A form action.
All templates must have at least one action, which is rendered as the
button that submits the form.
"""
class Meta:
unique_together = ('template', 'label', )
ordering = ('id', )
template = models.ForeignKey(FleetingTemplate,
on_delete=models.CASCADE,
related_name='actions')
label = models.CharField(max_length=30)
def __str__(self):
return self.label
# validations = models.ManyToThisField(FleetingValidation)
# choices = models.ManyToThisField(FleetingChoice)
[docs]class FleetingChoice(models.Model):
"""A choice field entry."""
class Meta:
unique_together = (('form_control', 'value'),
('form_control', 'text'), )
form_control = models.ForeignKey(FleetingFormControl,
on_delete=models.CASCADE,
related_name="choices")
value = models.CharField(max_length=30)
text = models.CharField(max_length=120)
def __str__(self):
return "{}: ({}, {})".format(
self.form_control.name, self.value, self.text)
[docs]class FleetingValidation(models.Model):
"""A custom field validation."""
class Meta:
unique_together = ('type', 'form_control', )
VALIDATION_TYPE_REGEX = 'regex'
VALIDATION_TYPE_MIN_LENGTH = 'min-length'
VALIDATION_TYPE_MAX_LENGTH = 'max-length'
VALIDATION_TYPE_RANGE_LENGTH = 'range-length'
VALIDATION_TYPE_MIN = 'min'
VALIDATION_TYPE_MAX = 'max'
VALIDATION_TYPE_RANGE = 'range'
VALIDATION_TYPES = (
(VALIDATION_TYPE_REGEX, 'Regular Expression'),
(VALIDATION_TYPE_MIN_LENGTH, 'Minimum Length'),
(VALIDATION_TYPE_MAX_LENGTH, 'Maximum Length'),
(VALIDATION_TYPE_RANGE_LENGTH, 'Length Range'),
(VALIDATION_TYPE_MIN, 'Minimum Value'),
(VALIDATION_TYPE_MAX, 'Maximum Value'),
(VALIDATION_TYPE_RANGE, 'Value Range'),
)
type = models.CharField(max_length=16, choices=VALIDATION_TYPES)
form_control = models.ForeignKey(FleetingFormControl,
on_delete=models.CASCADE,
related_name='validations')
params = JSONField(default=dict)
message = models.CharField(max_length=128)
def __str__(self):
return "{} {} [{}]".format(
self.type, self.params, self.form_control)
[docs]class FleetingAuditEntry(models.Model):
"""Fleeting Audit Entries track usage without personal details."""
code = models.CharField(unique=True,
max_length=settings.FLEETING_CODE_LENGTH)
namespace = models.ForeignKey(FleetingNamespace,
related_name='audit_entries',
on_delete=models.CASCADE)
auth = models.CharField(max_length=16,
choices=FleetingAuth.AUTH_TYPES,
default=FleetingAuth.AUTH_TYPE_NONE,)
status = models.CharField(max_length=max(
[len(s[0])
for s in FleetingForm.FORM_STATUSES]),
choices=FleetingForm.FORM_STATUSES,
default=FleetingForm.FORM_STATUS_CREATED)
template = models.CharField(max_length=max(
[len(s[0])
for s in FleetingTemplate.TEMPLATE_TYPES]),
choices=FleetingTemplate.TEMPLATE_TYPES,
default=FleetingTemplate.TEMPLATE_TYPE_GENERIC)
created_on = models.DateTimeField(null=False)
opened_on = models.DateTimeField(null=True)
completed_on = models.DateTimeField(null=True)
[docs] def delete(self, *args, **kwargs):
raise FleetingDeletionError(self.__class__.__name__, self.pk)
def __str__(self):
return "{} - {}".format(self.code, self.status)