Source code for fleetingform.serializers

from collections import OrderedDict
from datetime import timedelta
import bleach

from django.conf import settings
from django.utils import timezone

from rest_framework import serializers
from drf_writable_nested.serializers import WritableNestedModelSerializer

if settings.TWILIO_LOOKUP_VALIDATION:  # pragma: no cover
    from twilio.rest import Client as TwilioClient
    from twilio.base.exceptions import TwilioRestException

from fleetingform.models import (FleetingForm,
                                 FleetingNamespace,
                                 FleetingWebhook,
                                 FleetingTemplate,
                                 FleetingAuth,
                                 FleetingUser,
                                 FleetingAction,
                                 FleetingFormControl,
                                 FleetingChoice,
                                 FleetingValidation)

from fleetingform.errors import FleetingValidationError
from fleetingform.template_helpers import (FleetingAppParamsValidator,
                                           FleetingAuthValidator, )
from fleetingform.lib.form_control_validators import (
        DecimalFormControlParamsValidator,
        EmptyFormControlParamsValidator, )

ALLOWED_HTML_TAGS = ('a', 'abbr', 'acronym', 'b', 'blockquote', 'code', 'em',
                     'i', 'li', 'ol', 'strong', 'ul', 'p', 'br', 'table',
                     'thead', 'tbody', 'tfoot', 'th', 'td', 'tr', )


class FleetingWebhookSerializer(serializers.ModelSerializer):
    id = serializers.ReadOnlyField()
    event = serializers.ChoiceField(choices=FleetingWebhook.WEBHOOK_EVENTS)

    class Meta:
        model = FleetingWebhook
        fields = ['id', 'namespace', 'url', 'token', 'name', 'event']


class FleetingNamespaceSerializer(serializers.ModelSerializer):
    id = serializers.ReadOnlyField()
    user = serializers.ReadOnlyField(source='user.username')
    usage = serializers.SerializerMethodField()
    token = serializers.ReadOnlyField()

    def get_usage(self, namespace):
        this_month = timezone.now()
        last_month = this_month.replace(day=1) - timedelta(days=1)
        return {this_month.strftime("%Y-%m"): namespace.forms_this_month,
                last_month.strftime("%Y-%m"): namespace.forms_last_month}

    def to_representation(self, instance):
        """Exclude token unless user is authenticated."""
        result = super().to_representation(instance)
        request = self.context.get('request')
        if request and request.user:
            if request.auth == str(instance.token):
                del result['token']
        else:
            del result['token']

        return result

    class Meta:
        model = FleetingNamespace
        fields = ['id', 'user', 'subdomain', 'url_shortener',
                  'retention', 'usage', 'support_email', 'token',
                  'soft_limit', 'hard_limit', 'logo', 'style', ]


class FleetingActionSerializer(serializers.ModelSerializer):

    class Meta:
        model = FleetingAction
        fields = ['label']


class FleetingChoiceSerializer(serializers.ModelSerializer):

    class Meta:
        model = FleetingChoice
        fields = ['value', 'text']


class FleetingValidationSerializer(serializers.ModelSerializer):
    type = serializers.ChoiceField(choices=FleetingValidation.VALIDATION_TYPES)

    class Meta:
        model = FleetingValidation
        fields = ['type', 'params', 'message']


class FleetingFormControlSerializer(WritableNestedModelSerializer):
    # id = serializers.ReadOnlyField()
    choices = FleetingChoiceSerializer(
                        read_only=False, many=True, required=False)
    validations = FleetingValidationSerializer(
                        read_only=False, many=True, required=False)

    def validate(self, data):
        """Validate that the control has a valid combination of options."""
        if data.get('type') != FleetingFormControl.FIELD_TYPE_CHOICE:
            if data.get('choices'):
                raise serializers.ValidationError({
                    data.get('name'): ("Choices can only be provided "
                                       "for 'choice' fields.")
                    })
        else:
            if not data.get('choices'):
                raise serializers.ValidationError({
                    data.get('name'): ("At least one choice must be provided "
                                       "for 'choice' fields.")
                    })

        if data.get('type') == FleetingFormControl.FIELD_TYPE_DECIMAL:
            params_validator = DecimalFormControlParamsValidator(
                    data.get('params', {}))
        else:
            params_validator = EmptyFormControlParamsValidator(
                    data.get('params', {}))
        data['params'] = params_validator.validate()

        return data

    def to_representation(self, instance):
        """Exclude empty or null keys from serialized responses."""
        result = super().to_representation(instance)
        if instance.type != "choice":
            del result['choices']

        return result

    class Meta:
        model = FleetingFormControl
        fields = ['type', 'name', 'help_text', 'label',
                  'validations', 'choices', 'required',
                  'initial', 'disabled', 'params', 'hidden', ]


class FleetingTemplateSerializer(WritableNestedModelSerializer):
    type = serializers.ChoiceField(
                           choices=FleetingTemplate.TEMPLATE_TYPES,
                           default=FleetingTemplate.TEMPLATE_TYPE_GENERIC)
    actions = FleetingActionSerializer(read_only=False, many=True)
    form_controls = FleetingFormControlSerializer(
            read_only=False, many=True, required=False)

    def validate(self, data):
        """Validate that the template has a valid combination of options."""
        # Per-template validator for optional params
        template_helper = FleetingTemplate.template_helper_class_for(
                            data['type'])
        template_helper().validate_params(data.get('params'))
        return data

    def validate_actions(self, actions):
        """Validate the template actions."""
        if not len(actions):
            raise serializers.ValidationError(
                    "At least one action is required.")

    def validate_content(self, content):
        """Bleach any inbound HTML in the form content."""
        return bleach.clean(content, tags=ALLOWED_HTML_TAGS, strip=True)

    def validate_title(self, title):
        """Bleach any inbound HTML in the form title."""
        clean_title = bleach.clean(
                title, tags=ALLOWED_HTML_TAGS, strip=True)
        if len(clean_title) > 200:
            raise serializers.ValidationError(
                    "Title must be 120/200 or fewer characters before/after "
                    "HTML cleaning. Current title length {}/{} "
                    "before/after cleaning.".format(
                        len(title), len(clean_title)))
        return clean_title

    class Meta:
        model = FleetingTemplate
        fields = ['type', 'title', 'content', 'content_type',
                  'form_controls', 'params', 'actions']


class FleetingUserSerializer(serializers.ModelSerializer):
    opened_on = serializers.ReadOnlyField()

    def validate_password(self, data):
        """Validate an inbound password.

        Passwords may only come in two formats:
        - a string prefixed with ``plain:``
        - PHC string format (https://github.com/P-H-C/phc-string-format/blob/
                             master/phc-sf-spec.md)

        Plain text passwords are immediately hashed before being stored.
        PHC strings are checked for formatting and valid hashing type.
        """
        if not data:
            return data

        if data.startswith('plain:'):
            _, password = data.split(':', 1)
            if not password or \
                    len(password) < settings.FLEETING_PASSWORD_MIN_CHARS:
                raise serializers.ValidationError(
                        "plain passwords must have at least {} "
                        "characters.".format(
                            settings.FLEETING_PASSWORD_MIN_CHARS))
            data = settings.FLEETING_DEFAULT_HASHER.hash(password)
        else:
            try:
                _, algo, it, salt, hsh = data.split("$")
            except (ValueError, IndexError):
                raise serializers.ValidationError(
                        "hash format not supported, prefix plaintext passwords"
                        " with 'plain:' or provide a password in the modulo "
                        "crypto format.")

            if algo not in settings.FLEETING_HASHERS:
                raise serializers.ValidationError(
                        "hashing algorithm not supported, select from "
                        "{}.".format(", ".join(
                            settings.FLEETING_HASHERS)))
        return data

    def validate_phone(self, data):
        """If enabled, use the twilio lookup API to validate a phone number."""
        # twilio requires prod keys to test - can just be disabled.
        if settings.TWILIO_LOOKUP_VALIDATION:  # pragma: no cover
            try:
                tc = TwilioClient(settings.TWILIO_ACCOUNT_SID,
                                  settings.TWILIO_ACCOUNT_TOKEN)
                tc.lookups.phone_numbers(data).fetch(type=['carrier'])
            except TwilioRestException:
                raise serializers.ValidationError(
                        "phone number verification failed for {}.".format(
                            data))
        return data

    def to_representation(self, instance):
        """Exclude empty or null keys and hide password hashes."""
        result = super().to_representation(instance)
        if instance.auth.requires_password:
            result['password'] = "encrypted"
        else:
            del result['password']

        return result

    class Meta:
        model = FleetingUser
        fields = ['username', 'password', 'email', 'phone',
                  'opened_on', ]


class FleetingAuthSerializer(WritableNestedModelSerializer):
    REQUIRED_FORM_CONTROLS = {'username', 'password', }
    REQUIRED_FORM_CONTROL_DEFAULTS = {
            'username': {
                    'type': 'text',
                    'name': 'username',
                    'label': 'Username',
                    'required': True,
            },
            'password': {
                    'type': 'text',
                    'name': 'password',
                    'label': 'Password',
                    'required': True,
                    'validations': [
                        {
                            'type': 'max-length',
                            'params': {'max': 32},
                            'message': 'too long.'
                        }
                     ]
            }
        }

    # id = serializers.ReadOnlyField()
    users = FleetingUserSerializer(read_only=False, many=True, required=False)
    form_controls = FleetingFormControlSerializer(read_only=False,
                                                  many=True,
                                                  required=False)
    type = serializers.ChoiceField(choices=FleetingAuth.AUTH_TYPES,
                                   default=FleetingAuth.AUTH_TYPE_NONE)

    def validate(self, data):
        """Validate correct user arguments have been passed given the type"""
        data = FleetingAuthValidator(data['type']).validate(data)
        return data

    def create(self, validated_data):
        """Overridden create to add any required form controls."""
        form_control_names = {fc['name'] for fc in validated_data.get(
                                                    'form_controls', [])}
        missing_controls = self.REQUIRED_FORM_CONTROLS - form_control_names

        auth = super().create(validated_data)

        if auth.type is not FleetingAuth.AUTH_TYPE_NONE:
            if not auth.title:
                auth.title = FleetingAuth.AUTH_DEFAULT_TITLE
            if not auth.content:
                auth.content = FleetingAuth.AUTH_DEFAULT_CONTENT
            auth.save()

            if 'user' in auth.type and 'username' in missing_controls:
                fcs = FleetingFormControlSerializer(
                        data=self.REQUIRED_FORM_CONTROL_DEFAULTS['username'])
                if fcs.is_valid(raise_exception=True):  # pragma: no branch
                    fcs.save(form=auth)
            if ('pass' in auth.type or 'otp' in auth.type) and \
                    'password' in missing_controls:
                fcs = FleetingFormControlSerializer(
                        data=self.REQUIRED_FORM_CONTROL_DEFAULTS['password'])
                if fcs.is_valid(raise_exception=True):  # pragma: no branch
                    fcs.save(form=auth)

        return auth

    def to_representation(self, instance):
        """Exclude empty or null keys from serialized responses."""
        result = super().to_representation(instance)
        return OrderedDict([(key, result[key])
                           for key in result
                           if result[key]])

    class Meta:
        model = FleetingAuth
        fields = ['type', 'title', 'content',
                  'form_controls', 'users', ]


class FleetingFormSerializer(WritableNestedModelSerializer):
    id = serializers.ReadOnlyField()
    code = serializers.ReadOnlyField()
    url = serializers.ReadOnlyField()
    short_url = serializers.ReadOnlyField()
    status = serializers.ReadOnlyField()
    created_on = serializers.ReadOnlyField()
    opened_on = serializers.ReadOnlyField()
    completed_on = serializers.ReadOnlyField()
    expires_on = serializers.ReadOnlyField()
    result = serializers.ReadOnlyField()
    template = FleetingTemplateSerializer(read_only=False, many=False)
    auth = FleetingAuthSerializer(read_only=False, many=False, required=False)

    class Meta:
        model = FleetingForm
        fields = ['id', 'code', 'template', 'app', 'auth',
                  'url', 'short_url', 'status',
                  'created_on', 'opened_on', 'completed_on',
                  'expires_on', 'result', ]

    def validate_namespace(self, val):
        """Populate the namespace from the request for new creates."""
        if not val:
            r = self.context.get('request')
            if r and hasattr(r, 'namespace'):
                return r.namespace
        return val

    def validate_app(self, val):
        """Validate that the app parameters are within bounds."""
        try:
            FleetingAppParamsValidator().validate(val)
        except FleetingValidationError as e:
            raise serializers.ValidationError({e.field: str(e)})
        return val

    def validate_auth(self, data):
        """Populate default auth type for new creates without type."""
        if not data:
            data = {'type': FleetingAuth.AUTH_TYPE_NONE}
        return data

    def create(self, validated_data):
        form = super().create(validated_data)
        if not form.auth:
            auth = FleetingAuth(type=FleetingAuth.AUTH_TYPE_NONE)
            auth.save()
            form.auth = auth
            form.save()
        return form