Plain

  1import ipaddress
  2import math
  3import re
  4from pathlib import Path
  5from urllib.parse import urlsplit, urlunsplit
  6
  7from plain.exceptions import ValidationError
  8from plain.utils.deconstruct import deconstructible
  9from plain.utils.encoding import punycode
 10from plain.utils.ipv6 import is_valid_ipv6_address
 11from plain.utils.regex_helper import _lazy_re_compile
 12from plain.utils.text import pluralize_lazy
 13
 14# These values, if given to validate(), will trigger the self.required check.
 15EMPTY_VALUES = (None, "", [], (), {})
 16
 17
 18@deconstructible
 19class RegexValidator:
 20    regex = ""
 21    message = "Enter a valid value."
 22    code = "invalid"
 23    inverse_match = False
 24    flags = 0
 25
 26    def __init__(
 27        self, regex=None, message=None, code=None, inverse_match=None, flags=None
 28    ):
 29        if regex is not None:
 30            self.regex = regex
 31        if message is not None:
 32            self.message = message
 33        if code is not None:
 34            self.code = code
 35        if inverse_match is not None:
 36            self.inverse_match = inverse_match
 37        if flags is not None:
 38            self.flags = flags
 39        if self.flags and not isinstance(self.regex, str):
 40            raise TypeError(
 41                "If the flags are set, regex must be a regular expression string."
 42            )
 43
 44        self.regex = _lazy_re_compile(self.regex, self.flags)
 45
 46    def __call__(self, value):
 47        """
 48        Validate that the input contains (or does *not* contain, if
 49        inverse_match is True) a match for the regular expression.
 50        """
 51        regex_matches = self.regex.search(str(value))
 52        invalid_input = regex_matches if self.inverse_match else not regex_matches
 53        if invalid_input:
 54            raise ValidationError(self.message, code=self.code, params={"value": value})
 55
 56    def __eq__(self, other):
 57        return (
 58            isinstance(other, RegexValidator)
 59            and self.regex.pattern == other.regex.pattern
 60            and self.regex.flags == other.regex.flags
 61            and (self.message == other.message)
 62            and (self.code == other.code)
 63            and (self.inverse_match == other.inverse_match)
 64        )
 65
 66
 67@deconstructible
 68class URLValidator(RegexValidator):
 69    ul = "\u00a1-\uffff"  # Unicode letters range (must not be a raw string).
 70
 71    # IP patterns
 72    ipv4_re = (
 73        r"(?:0|25[0-5]|2[0-4][0-9]|1[0-9]?[0-9]?|[1-9][0-9]?)"
 74        r"(?:\.(?:0|25[0-5]|2[0-4][0-9]|1[0-9]?[0-9]?|[1-9][0-9]?)){3}"
 75    )
 76    ipv6_re = r"\[[0-9a-f:.]+\]"  # (simple regex, validated later)
 77
 78    # Host patterns
 79    hostname_re = (
 80        r"[a-z" + ul + r"0-9](?:[a-z" + ul + r"0-9-]{0,61}[a-z" + ul + r"0-9])?"
 81    )
 82    # Max length for domain name labels is 63 characters per RFC 1034 sec. 3.1
 83    domain_re = r"(?:\.(?!-)[a-z" + ul + r"0-9-]{1,63}(?<!-))*"
 84    tld_re = (
 85        r"\."  # dot
 86        r"(?!-)"  # can't start with a dash
 87        r"(?:[a-z" + ul + "-]{2,63}"  # domain label
 88        r"|xn--[a-z0-9]{1,59})"  # or punycode label
 89        r"(?<!-)"  # can't end with a dash
 90        r"\.?"  # may have a trailing dot
 91    )
 92    host_re = "(" + hostname_re + domain_re + tld_re + "|localhost)"
 93
 94    regex = _lazy_re_compile(
 95        r"^(?:[a-z0-9.+-]*)://"  # scheme is validated separately
 96        r"(?:[^\s:@/]+(?::[^\s:@/]*)?@)?"  # user:pass authentication
 97        r"(?:" + ipv4_re + "|" + ipv6_re + "|" + host_re + ")"
 98        r"(?::[0-9]{1,5})?"  # port
 99        r"(?:[/?#][^\s]*)?"  # resource path
100        r"\Z",
101        re.IGNORECASE,
102    )
103    message = "Enter a valid URL."
104    schemes = ["http", "https", "ftp", "ftps"]
105    unsafe_chars = frozenset("\t\r\n")
106
107    def __init__(self, schemes=None, **kwargs):
108        super().__init__(**kwargs)
109        if schemes is not None:
110            self.schemes = schemes
111
112    def __call__(self, value):
113        if not isinstance(value, str):
114            raise ValidationError(self.message, code=self.code, params={"value": value})
115        if self.unsafe_chars.intersection(value):
116            raise ValidationError(self.message, code=self.code, params={"value": value})
117        # Check if the scheme is valid.
118        scheme = value.split("://")[0].lower()
119        if scheme not in self.schemes:
120            raise ValidationError(self.message, code=self.code, params={"value": value})
121
122        # Then check full URL
123        try:
124            splitted_url = urlsplit(value)
125        except ValueError:
126            raise ValidationError(self.message, code=self.code, params={"value": value})
127        try:
128            super().__call__(value)
129        except ValidationError as e:
130            # Trivial case failed. Try for possible IDN domain
131            if value:
132                scheme, netloc, path, query, fragment = splitted_url
133                try:
134                    netloc = punycode(netloc)  # IDN -> ACE
135                except UnicodeError:  # invalid domain part
136                    raise e
137                url = urlunsplit((scheme, netloc, path, query, fragment))
138                super().__call__(url)
139            else:
140                raise
141        else:
142            # Now verify IPv6 in the netloc part
143            host_match = re.search(r"^\[(.+)\](?::[0-9]{1,5})?$", splitted_url.netloc)
144            if host_match:
145                potential_ip = host_match[1]
146                try:
147                    validate_ipv6_address(potential_ip)
148                except ValidationError:
149                    raise ValidationError(
150                        self.message, code=self.code, params={"value": value}
151                    )
152
153        # The maximum length of a full host name is 253 characters per RFC 1034
154        # section 3.1. It's defined to be 255 bytes or less, but this includes
155        # one byte for the length of the name and one byte for the trailing dot
156        # that's used to indicate absolute names in DNS.
157        if splitted_url.hostname is None or len(splitted_url.hostname) > 253:
158            raise ValidationError(self.message, code=self.code, params={"value": value})
159
160
161integer_validator = RegexValidator(
162    _lazy_re_compile(r"^-?\d+\Z"),
163    message="Enter a valid integer.",
164    code="invalid",
165)
166
167
168def validate_integer(value):
169    return integer_validator(value)
170
171
172@deconstructible
173class EmailValidator:
174    message = "Enter a valid email address."
175    code = "invalid"
176    user_regex = _lazy_re_compile(
177        # dot-atom
178        r"(^[-!#$%&'*+/=?^_`{}|~0-9A-Z]+(\.[-!#$%&'*+/=?^_`{}|~0-9A-Z]+)*\Z"
179        # quoted-string
180        r'|^"([\001-\010\013\014\016-\037!#-\[\]-\177]|\\[\001-\011\013\014\016-\177])'
181        r'*"\Z)',
182        re.IGNORECASE,
183    )
184    domain_regex = _lazy_re_compile(
185        # max length for domain name labels is 63 characters per RFC 1034
186        r"((?:[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?\.)+)(?:[A-Z0-9-]{2,63}(?<!-))\Z",
187        re.IGNORECASE,
188    )
189    literal_regex = _lazy_re_compile(
190        # literal form, ipv4 or ipv6 address (SMTP 4.1.3)
191        r"\[([A-F0-9:.]+)\]\Z",
192        re.IGNORECASE,
193    )
194    domain_allowlist = ["localhost"]
195
196    def __init__(self, message=None, code=None, allowlist=None):
197        if message is not None:
198            self.message = message
199        if code is not None:
200            self.code = code
201        if allowlist is not None:
202            self.domain_allowlist = allowlist
203
204    def __call__(self, value):
205        if not value or "@" not in value:
206            raise ValidationError(self.message, code=self.code, params={"value": value})
207
208        user_part, domain_part = value.rsplit("@", 1)
209
210        if not self.user_regex.match(user_part):
211            raise ValidationError(self.message, code=self.code, params={"value": value})
212
213        if domain_part not in self.domain_allowlist and not self.validate_domain_part(
214            domain_part
215        ):
216            # Try for possible IDN domain-part
217            try:
218                domain_part = punycode(domain_part)
219            except UnicodeError:
220                pass
221            else:
222                if self.validate_domain_part(domain_part):
223                    return
224            raise ValidationError(self.message, code=self.code, params={"value": value})
225
226    def validate_domain_part(self, domain_part):
227        if self.domain_regex.match(domain_part):
228            return True
229
230        literal_match = self.literal_regex.match(domain_part)
231        if literal_match:
232            ip_address = literal_match[1]
233            try:
234                validate_ipv46_address(ip_address)
235                return True
236            except ValidationError:
237                pass
238        return False
239
240    def __eq__(self, other):
241        return (
242            isinstance(other, EmailValidator)
243            and (self.domain_allowlist == other.domain_allowlist)
244            and (self.message == other.message)
245            and (self.code == other.code)
246        )
247
248
249validate_email = EmailValidator()
250
251slug_re = _lazy_re_compile(r"^[-a-zA-Z0-9_]+\Z")
252validate_slug = RegexValidator(
253    slug_re,
254    # Translators: "letters" means latin letters: a-z and A-Z.
255    "Enter a valid “slug” consisting of letters, numbers, underscores or hyphens.",
256    "invalid",
257)
258
259slug_unicode_re = _lazy_re_compile(r"^[-\w]+\Z")
260validate_unicode_slug = RegexValidator(
261    slug_unicode_re,
262    "Enter a valid “slug” consisting of Unicode letters, numbers, underscores, or hyphens."
263    "invalid",
264)
265
266
267def validate_ipv4_address(value):
268    try:
269        ipaddress.IPv4Address(value)
270    except ValueError:
271        raise ValidationError(
272            "Enter a valid IPv4 address.", code="invalid", params={"value": value}
273        )
274
275
276def validate_ipv6_address(value):
277    if not is_valid_ipv6_address(value):
278        raise ValidationError(
279            "Enter a valid IPv6 address.", code="invalid", params={"value": value}
280        )
281
282
283def validate_ipv46_address(value):
284    try:
285        validate_ipv4_address(value)
286    except ValidationError:
287        try:
288            validate_ipv6_address(value)
289        except ValidationError:
290            raise ValidationError(
291                "Enter a valid IPv4 or IPv6 address.",
292                code="invalid",
293                params={"value": value},
294            )
295
296
297ip_address_validator_map = {
298    "both": ([validate_ipv46_address], "Enter a valid IPv4 or IPv6 address."),
299    "ipv4": ([validate_ipv4_address], "Enter a valid IPv4 address."),
300    "ipv6": ([validate_ipv6_address], "Enter a valid IPv6 address."),
301}
302
303
304def ip_address_validators(protocol, unpack_ipv4):
305    """
306    Depending on the given parameters, return the appropriate validators for
307    the GenericIPAddressField.
308    """
309    if protocol != "both" and unpack_ipv4:
310        raise ValueError(
311            "You can only use `unpack_ipv4` if `protocol` is set to 'both'"
312        )
313    try:
314        return ip_address_validator_map[protocol.lower()]
315    except KeyError:
316        raise ValueError(
317            f"The protocol '{protocol}' is unknown. Supported: {list(ip_address_validator_map)}"
318        )
319
320
321def int_list_validator(sep=",", message=None, code="invalid", allow_negative=False):
322    regexp = _lazy_re_compile(
323        r"^{neg}\d+(?:{sep}{neg}\d+)*\Z".format(
324            neg="(-)?" if allow_negative else "",
325            sep=re.escape(sep),
326        )
327    )
328    return RegexValidator(regexp, message=message, code=code)
329
330
331validate_comma_separated_integer_list = int_list_validator(
332    message="Enter only digits separated by commas.",
333)
334
335
336@deconstructible
337class BaseValidator:
338    message = "Ensure this value is %(limit_value)s (it is %(show_value)s)."
339    code = "limit_value"
340
341    def __init__(self, limit_value, message=None):
342        self.limit_value = limit_value
343        if message:
344            self.message = message
345
346    def __call__(self, value):
347        cleaned = self.clean(value)
348        limit_value = (
349            self.limit_value() if callable(self.limit_value) else self.limit_value
350        )
351        params = {"limit_value": limit_value, "show_value": cleaned, "value": value}
352        if self.compare(cleaned, limit_value):
353            raise ValidationError(self.message, code=self.code, params=params)
354
355    def __eq__(self, other):
356        if not isinstance(other, self.__class__):
357            return NotImplemented
358        return (
359            self.limit_value == other.limit_value
360            and self.message == other.message
361            and self.code == other.code
362        )
363
364    def compare(self, a, b):
365        return a is not b
366
367    def clean(self, x):
368        return x
369
370
371@deconstructible
372class MaxValueValidator(BaseValidator):
373    message = "Ensure this value is less than or equal to %(limit_value)s."
374    code = "max_value"
375
376    def compare(self, a, b):
377        return a > b
378
379
380@deconstructible
381class MinValueValidator(BaseValidator):
382    message = "Ensure this value is greater than or equal to %(limit_value)s."
383    code = "min_value"
384
385    def compare(self, a, b):
386        return a < b
387
388
389@deconstructible
390class StepValueValidator(BaseValidator):
391    message = "Ensure this value is a multiple of step size %(limit_value)s."
392    code = "step_size"
393
394    def compare(self, a, b):
395        return not math.isclose(math.remainder(a, b), 0, abs_tol=1e-9)
396
397
398@deconstructible
399class MinLengthValidator(BaseValidator):
400    message = pluralize_lazy(
401        "Ensure this value has at least %(limit_value)d character (it has "
402        "%(show_value)d).",
403        "Ensure this value has at least %(limit_value)d characters (it has "
404        "%(show_value)d).",
405        "limit_value",
406    )
407    code = "min_length"
408
409    def compare(self, a, b):
410        return a < b
411
412    def clean(self, x):
413        return len(x)
414
415
416@deconstructible
417class MaxLengthValidator(BaseValidator):
418    message = pluralize_lazy(
419        "Ensure this value has at most %(limit_value)d character (it has "
420        "%(show_value)d).",
421        "Ensure this value has at most %(limit_value)d characters (it has "
422        "%(show_value)d).",
423        "limit_value",
424    )
425    code = "max_length"
426
427    def compare(self, a, b):
428        return a > b
429
430    def clean(self, x):
431        return len(x)
432
433
434@deconstructible
435class DecimalValidator:
436    """
437    Validate that the input does not exceed the maximum number of digits
438    expected, otherwise raise ValidationError.
439    """
440
441    messages = {
442        "invalid": "Enter a number.",
443        "max_digits": pluralize_lazy(
444            "Ensure that there are no more than %(max)s digit in total.",
445            "Ensure that there are no more than %(max)s digits in total.",
446            "max",
447        ),
448        "max_decimal_places": pluralize_lazy(
449            "Ensure that there are no more than %(max)s decimal place.",
450            "Ensure that there are no more than %(max)s decimal places.",
451            "max",
452        ),
453        "max_whole_digits": pluralize_lazy(
454            "Ensure that there are no more than %(max)s digit before the decimal "
455            "point.",
456            "Ensure that there are no more than %(max)s digits before the decimal "
457            "point.",
458            "max",
459        ),
460    }
461
462    def __init__(self, max_digits, decimal_places):
463        self.max_digits = max_digits
464        self.decimal_places = decimal_places
465
466    def __call__(self, value):
467        digit_tuple, exponent = value.as_tuple()[1:]
468        if exponent in {"F", "n", "N"}:
469            raise ValidationError(
470                self.messages["invalid"], code="invalid", params={"value": value}
471            )
472        if exponent >= 0:
473            digits = len(digit_tuple)
474            if digit_tuple != (0,):
475                # A positive exponent adds that many trailing zeros.
476                digits += exponent
477            decimals = 0
478        else:
479            # If the absolute value of the negative exponent is larger than the
480            # number of digits, then it's the same as the number of digits,
481            # because it'll consume all of the digits in digit_tuple and then
482            # add abs(exponent) - len(digit_tuple) leading zeros after the
483            # decimal point.
484            if abs(exponent) > len(digit_tuple):
485                digits = decimals = abs(exponent)
486            else:
487                digits = len(digit_tuple)
488                decimals = abs(exponent)
489        whole_digits = digits - decimals
490
491        if self.max_digits is not None and digits > self.max_digits:
492            raise ValidationError(
493                self.messages["max_digits"],
494                code="max_digits",
495                params={"max": self.max_digits, "value": value},
496            )
497        if self.decimal_places is not None and decimals > self.decimal_places:
498            raise ValidationError(
499                self.messages["max_decimal_places"],
500                code="max_decimal_places",
501                params={"max": self.decimal_places, "value": value},
502            )
503        if (
504            self.max_digits is not None
505            and self.decimal_places is not None
506            and whole_digits > (self.max_digits - self.decimal_places)
507        ):
508            raise ValidationError(
509                self.messages["max_whole_digits"],
510                code="max_whole_digits",
511                params={"max": (self.max_digits - self.decimal_places), "value": value},
512            )
513
514    def __eq__(self, other):
515        return (
516            isinstance(other, self.__class__)
517            and self.max_digits == other.max_digits
518            and self.decimal_places == other.decimal_places
519        )
520
521
522@deconstructible
523class FileExtensionValidator:
524    message = "File extension “%(extension)s” is not allowed. Allowed extensions are: %(allowed_extensions)s."
525    code = "invalid_extension"
526
527    def __init__(self, allowed_extensions=None, message=None, code=None):
528        if allowed_extensions is not None:
529            allowed_extensions = [
530                allowed_extension.lower() for allowed_extension in allowed_extensions
531            ]
532        self.allowed_extensions = allowed_extensions
533        if message is not None:
534            self.message = message
535        if code is not None:
536            self.code = code
537
538    def __call__(self, value):
539        extension = Path(value.name).suffix[1:].lower()
540        if (
541            self.allowed_extensions is not None
542            and extension not in self.allowed_extensions
543        ):
544            raise ValidationError(
545                self.message,
546                code=self.code,
547                params={
548                    "extension": extension,
549                    "allowed_extensions": ", ".join(self.allowed_extensions),
550                    "value": value,
551                },
552            )
553
554    def __eq__(self, other):
555        return (
556            isinstance(other, self.__class__)
557            and self.allowed_extensions == other.allowed_extensions
558            and self.message == other.message
559            and self.code == other.code
560        )
561
562
563def get_available_image_extensions():
564    try:
565        from PIL import Image
566    except ImportError:
567        return []
568    else:
569        Image.init()
570        return [ext.lower()[1:] for ext in Image.EXTENSION]
571
572
573def validate_image_file_extension(value):
574    return FileExtensionValidator(allowed_extensions=get_available_image_extensions())(
575        value
576    )
577
578
579@deconstructible
580class ProhibitNullCharactersValidator:
581    """Validate that the string doesn't contain the null character."""
582
583    message = "Null characters are not allowed."
584    code = "null_characters_not_allowed"
585
586    def __init__(self, message=None, code=None):
587        if message is not None:
588            self.message = message
589        if code is not None:
590            self.code = code
591
592    def __call__(self, value):
593        if "\x00" in str(value):
594            raise ValidationError(self.message, code=self.code, params={"value": value})
595
596    def __eq__(self, other):
597        return (
598            isinstance(other, self.__class__)
599            and self.message == other.message
600            and self.code == other.code
601        )