# -*- coding: utf-8 -*-
"""
This module provide some helpers for advanced types parsing.
You can define you own parser using the same pattern:
.. code-block:: python
def my_type(value):
if not condition:
raise ValueError('This is not my type')
return parse(value)
# Swagger documentation
my_type.__schema__ = {'type': 'string', 'format': 'my-custom-format'}
The last line allows you to document properly the type in the Swagger documentation.
"""
from __future__ import unicode_literals
import re
import socket
from datetime import datetime, time, timedelta
from email.utils import parsedate_tz, mktime_tz
from six.moves.urllib.parse import urlparse
import aniso8601
import pytz
# Constants for upgrading date-based intervals to full datetimes.
START_OF_DAY = time(0, 0, 0, tzinfo=pytz.UTC)
END_OF_DAY = time(23, 59, 59, 999999, tzinfo=pytz.UTC)
netloc_regex = re.compile(
r"(?:(?P<auth>[^:@]+?(?::[^:@]*?)?)@)?" # basic auth
r"(?:"
r"(?P<localhost>localhost)|" # localhost...
r"(?P<ipv4>\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})|" # ...or ipv4
r"(?:\[?(?P<ipv6>[A-F0-9]*:[A-F0-9:]+)\]?)|" # ...or ipv6
r"(?P<domain>(?:[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?\.)+(?:[A-Z]{2,6}\.?|[A-Z0-9-]{2,}\.?))" # domain...
r")"
r"(?::(?P<port>\d+))?" # optional port
r"$",
re.IGNORECASE,
)
email_regex = re.compile(
r"^" "(?P<local>[^@]*[^@.])" r"@" r"(?P<server>[^@\.]+(?:\.[^@\.]+)*)" r"$",
re.IGNORECASE,
)
time_regex = re.compile(r"\d{2}:\d{2}")
[docs]def ipv4(value):
"""Validate an IPv4 address"""
try:
socket.inet_aton(value)
if value.count(".") == 3:
return value
except socket.error:
pass
raise ValueError("{0} is not a valid ipv4 address".format(value))
ipv4.__schema__ = {"type": "string", "format": "ipv4"}
[docs]def ipv6(value):
"""Validate an IPv6 address"""
try:
socket.inet_pton(socket.AF_INET6, value)
return value
except socket.error:
raise ValueError("{0} is not a valid ipv4 address".format(value))
ipv6.__schema__ = {"type": "string", "format": "ipv6"}
[docs]def ip(value):
"""Validate an IP address (both IPv4 and IPv6)"""
try:
return ipv4(value)
except ValueError:
pass
try:
return ipv6(value)
except ValueError:
raise ValueError("{0} is not a valid ip".format(value))
ip.__schema__ = {"type": "string", "format": "ip"}
[docs]class URL(object):
"""
Validate an URL.
Example::
parser = reqparse.RequestParser()
parser.add_argument('url', type=inputs.URL(schemes=['http', 'https']))
Input to the ``URL`` argument will be rejected
if it does not match an URL with specified constraints.
If ``check`` is True it will also be rejected if the domain does not exists.
:param bool check: Check the domain exists (perform a DNS resolution)
:param bool ip: Allow IP (both ipv4/ipv6) as domain
:param bool local: Allow localhost (both string or ip) as domain
:param bool port: Allow a port to be present
:param bool auth: Allow authentication to be present
:param list|tuple schemes: Restrict valid schemes to this list
:param list|tuple domains: Restrict valid domains to this list
:param list|tuple exclude: Exclude some domains
"""
def __init__(
self,
check=False,
ip=False,
local=False,
port=False,
auth=False,
schemes=None,
domains=None,
exclude=None,
):
self.check = check
self.ip = ip
self.local = local
self.port = port
self.auth = auth
self.schemes = schemes
self.domains = domains
self.exclude = exclude
def error(self, value, details=None):
msg = "{0} is not a valid URL"
if details:
msg = ". ".join((msg, details))
raise ValueError(msg.format(value))
def __call__(self, value):
parsed = urlparse(value)
netloc_match = netloc_regex.match(parsed.netloc)
if not all((parsed.scheme, parsed.netloc)):
if netloc_regex.match(
parsed.netloc or parsed.path.split("/", 1)[0].split("?", 1)[0]
):
self.error(value, "Did you mean: http://{0}")
self.error(value)
if parsed.scheme and self.schemes and parsed.scheme not in self.schemes:
self.error(value, "Protocol is not allowed")
if not netloc_match:
self.error(value)
data = netloc_match.groupdict()
if data["ipv4"] or data["ipv6"]:
if not self.ip:
self.error(value, "IP is not allowed")
else:
try:
ip(data["ipv4"] or data["ipv6"])
except ValueError as e:
self.error(value, str(e))
if not self.local:
if data["ipv4"] and data["ipv4"].startswith("127."):
self.error(value, "Localhost is not allowed")
elif data["ipv6"] == "::1":
self.error(value, "Localhost is not allowed")
if self.check:
pass
if data["auth"] and not self.auth:
self.error(value, "Authentication is not allowed")
if data["localhost"] and not self.local:
self.error(value, "Localhost is not allowed")
if data["port"]:
if not self.port:
self.error(value, "Custom port is not allowed")
else:
port = int(data["port"])
if not 0 < port < 65535:
self.error(value, "Port is out of range")
if data["domain"]:
if self.domains and data["domain"] not in self.domains:
self.error(value, "Domain is not allowed")
elif self.exclude and data["domain"] in self.exclude:
self.error(value, "Domain is not allowed")
if self.check:
try:
socket.getaddrinfo(data["domain"], None)
except socket.error:
self.error(value, "Domain does not exists")
return value
@property
def __schema__(self):
return {
"type": "string",
"format": "url",
}
#: Validate an URL
#:
#: Legacy validator, allows, auth, port, ip and local
#: Only allows schemes 'http', 'https', 'ftp' and 'ftps'
url = URL(
ip=True, auth=True, port=True, local=True, schemes=("http", "https", "ftp", "ftps")
)
[docs]class email(object):
"""
Validate an email.
Example::
parser = reqparse.RequestParser()
parser.add_argument('email', type=inputs.email(dns=True))
Input to the ``email`` argument will be rejected if it does not match an email
and if domain does not exists.
:param bool check: Check the domain exists (perform a DNS resolution)
:param bool ip: Allow IP (both ipv4/ipv6) as domain
:param bool local: Allow localhost (both string or ip) as domain
:param list|tuple domains: Restrict valid domains to this list
:param list|tuple exclude: Exclude some domains
"""
def __init__(self, check=False, ip=False, local=False, domains=None, exclude=None):
self.check = check
self.ip = ip
self.local = local
self.domains = domains
self.exclude = exclude
def error(self, value, msg=None):
msg = msg or "{0} is not a valid email"
raise ValueError(msg.format(value))
def is_ip(self, value):
try:
ip(value)
return True
except ValueError:
return False
def __call__(self, value):
match = email_regex.match(value)
if not match or ".." in value:
self.error(value)
server = match.group("server")
if self.check:
try:
socket.getaddrinfo(server, None)
except socket.error:
self.error(value)
if self.domains and server not in self.domains:
self.error(value, "{0} does not belong to the authorized domains")
if self.exclude and server in self.exclude:
self.error(value, "{0} belongs to a forbidden domain")
if not self.local and (
server in ("localhost", "::1") or server.startswith("127.")
):
self.error(value)
if self.is_ip(server) and not self.ip:
self.error(value)
return value
@property
def __schema__(self):
return {
"type": "string",
"format": "email",
}
[docs]class regex(object):
"""
Validate a string based on a regular expression.
Example::
parser = reqparse.RequestParser()
parser.add_argument('example', type=inputs.regex('^[0-9]+$'))
Input to the ``example`` argument will be rejected if it contains anything
but numbers.
:param str pattern: The regular expression the input must match
"""
def __init__(self, pattern):
self.pattern = pattern
self.re = re.compile(pattern)
def __call__(self, value):
if not self.re.search(value):
message = 'Value does not match pattern: "{0}"'.format(self.pattern)
raise ValueError(message)
return value
def __deepcopy__(self, memo):
return regex(self.pattern)
@property
def __schema__(self):
return {
"type": "string",
"pattern": self.pattern,
}
def _normalize_interval(start, end, value):
"""
Normalize datetime intervals.
Given a pair of datetime.date or datetime.datetime objects,
returns a 2-tuple of tz-aware UTC datetimes spanning the same interval.
For datetime.date objects, the returned interval starts at 00:00:00.0
on the first date and ends at 00:00:00.0 on the second.
Naive datetimes are upgraded to UTC.
Timezone-aware datetimes are normalized to the UTC tzdata.
Params:
- start: A date or datetime
- end: A date or datetime
"""
if not isinstance(start, datetime):
start = datetime.combine(start, START_OF_DAY)
end = datetime.combine(end, START_OF_DAY)
if start.tzinfo is None:
start = pytz.UTC.localize(start)
end = pytz.UTC.localize(end)
else:
start = start.astimezone(pytz.UTC)
end = end.astimezone(pytz.UTC)
return start, end
def _expand_datetime(start, value):
if not isinstance(start, datetime):
# Expand a single date object to be the interval spanning
# that entire day.
end = start + timedelta(days=1)
else:
# Expand a datetime based on the finest resolution provided
# in the original input string.
time = value.split("T")[1]
time_without_offset = re.sub("[+-].+", "", time)
num_separators = time_without_offset.count(":")
if num_separators == 0:
# Hour resolution
end = start + timedelta(hours=1)
elif num_separators == 1:
# Minute resolution:
end = start + timedelta(minutes=1)
else:
# Second resolution
end = start + timedelta(seconds=1)
return end
def _parse_interval(value):
"""
Do some nasty try/except voodoo to get some sort of datetime
object(s) out of the string.
"""
try:
return sorted(aniso8601.parse_interval(value))
except ValueError:
try:
return aniso8601.parse_datetime(value), None
except ValueError:
return aniso8601.parse_date(value), None
[docs]def iso8601interval(value, argument="argument"):
"""
Parses ISO 8601-formatted datetime intervals into tuples of datetimes.
Accepts both a single date(time) or a full interval using either start/end
or start/duration notation, with the following behavior:
- Intervals are defined as inclusive start, exclusive end
- Single datetimes are translated into the interval spanning the
largest resolution not specified in the input value, up to the day.
- The smallest accepted resolution is 1 second.
- All timezones are accepted as values; returned datetimes are
localized to UTC. Naive inputs and date inputs will are assumed UTC.
Examples::
"2013-01-01" -> datetime(2013, 1, 1), datetime(2013, 1, 2)
"2013-01-01T12" -> datetime(2013, 1, 1, 12), datetime(2013, 1, 1, 13)
"2013-01-01/2013-02-28" -> datetime(2013, 1, 1), datetime(2013, 2, 28)
"2013-01-01/P3D" -> datetime(2013, 1, 1), datetime(2013, 1, 4)
"2013-01-01T12:00/PT30M" -> datetime(2013, 1, 1, 12), datetime(2013, 1, 1, 12, 30)
"2013-01-01T06:00/2013-01-01T12:00" -> datetime(2013, 1, 1, 6), datetime(2013, 1, 1, 12)
:param str value: The ISO8601 date time as a string
:return: Two UTC datetimes, the start and the end of the specified interval
:rtype: A tuple (datetime, datetime)
:raises ValueError: if the interval is invalid.
"""
if not value:
raise ValueError("Expected a valid ISO8601 date/time interval.")
try:
start, end = _parse_interval(value)
if end is None:
end = _expand_datetime(start, value)
start, end = _normalize_interval(start, end, value)
except ValueError:
msg = (
"Invalid {arg}: {value}. {arg} must be a valid ISO8601 date/time interval."
)
raise ValueError(msg.format(arg=argument, value=value))
return start, end
iso8601interval.__schema__ = {"type": "string", "format": "iso8601-interval"}
[docs]def date(value):
"""Parse a valid looking date in the format YYYY-mm-dd"""
date = datetime.strptime(value, "%Y-%m-%d")
return date
date.__schema__ = {"type": "string", "format": "date"}
def _get_integer(value):
try:
return int(value)
except (TypeError, ValueError):
raise ValueError("{0} is not a valid integer".format(value))
[docs]def natural(value, argument="argument"):
"""Restrict input type to the natural numbers (0, 1, 2, 3...)"""
value = _get_integer(value)
if value < 0:
msg = "Invalid {arg}: {value}. {arg} must be a non-negative integer"
raise ValueError(msg.format(arg=argument, value=value))
return value
natural.__schema__ = {"type": "integer", "minimum": 0}
[docs]def positive(value, argument="argument"):
"""Restrict input type to the positive integers (1, 2, 3...)"""
value = _get_integer(value)
if value < 1:
msg = "Invalid {arg}: {value}. {arg} must be a positive integer"
raise ValueError(msg.format(arg=argument, value=value))
return value
positive.__schema__ = {"type": "integer", "minimum": 0, "exclusiveMinimum": True}
[docs]class int_range(object):
"""Restrict input to an integer in a range (inclusive)"""
def __init__(self, low, high, argument="argument"):
self.low = low
self.high = high
self.argument = argument
def __call__(self, value):
value = _get_integer(value)
if value < self.low or value > self.high:
msg = "Invalid {arg}: {val}. {arg} must be within the range {lo} - {hi}"
raise ValueError(
msg.format(arg=self.argument, val=value, lo=self.low, hi=self.high)
)
return value
@property
def __schema__(self):
return {
"type": "integer",
"minimum": self.low,
"maximum": self.high,
}
[docs]def boolean(value):
"""
Parse the string ``"true"`` or ``"false"`` as a boolean (case insensitive).
Also accepts ``"1"`` and ``"0"`` as ``True``/``False`` (respectively).
If the input is from the request JSON body, the type is already a native python boolean,
and will be passed through without further parsing.
:raises ValueError: if the boolean value is invalid
"""
if isinstance(value, bool):
return value
if value is None:
raise ValueError("boolean type must be non-null")
elif not value:
return False
value = str(value).lower()
if value in ("true", "1", "on",):
return True
if value in ("false", "0",):
return False
raise ValueError("Invalid literal for boolean(): {0}".format(value))
boolean.__schema__ = {"type": "boolean"}
[docs]def datetime_from_rfc822(value):
"""
Turns an RFC822 formatted date into a datetime object.
Example::
inputs.datetime_from_rfc822('Wed, 02 Oct 2002 08:00:00 EST')
:param str value: The RFC822-complying string to transform
:return: The parsed datetime
:rtype: datetime
:raises ValueError: if value is an invalid date literal
"""
raw = value
if not time_regex.search(value):
value = " ".join((value, "00:00:00"))
try:
timetuple = parsedate_tz(value)
timestamp = mktime_tz(timetuple)
if timetuple[-1] is None:
return datetime.fromtimestamp(timestamp).replace(tzinfo=pytz.utc)
else:
return datetime.fromtimestamp(timestamp, pytz.utc)
except Exception:
raise ValueError('Invalid date literal "{0}"'.format(raw))
[docs]def datetime_from_iso8601(value):
"""
Turns an ISO8601 formatted date into a datetime object.
Example::
inputs.datetime_from_iso8601("2012-01-01T23:30:00+02:00")
:param str value: The ISO8601-complying string to transform
:return: A datetime
:rtype: datetime
:raises ValueError: if value is an invalid date literal
"""
try:
try:
return aniso8601.parse_datetime(value)
except ValueError:
date = aniso8601.parse_date(value)
return datetime(date.year, date.month, date.day)
except Exception:
raise ValueError('Invalid date literal "{0}"'.format(value))
datetime_from_iso8601.__schema__ = {"type": "string", "format": "date-time"}
[docs]def date_from_iso8601(value):
"""
Turns an ISO8601 formatted date into a date object.
Example::
inputs.date_from_iso8601("2012-01-01")
:param str value: The ISO8601-complying string to transform
:return: A date
:rtype: date
:raises ValueError: if value is an invalid date literal
"""
return datetime_from_iso8601(value).date()
date_from_iso8601.__schema__ = {"type": "string", "format": "date"}
