~sircmpwn/core.sr.ht

core.sr.ht/srht/validation.py -rw-r--r-- 5.6 KiB
b695e020Drew DeVault Add "internal_anon" path for internal auth tokens 3 days ago
                                                                                
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
from markupsafe import Markup
from urllib import parse
from enum import Enum, IntEnum
import json

class ValidationError:
    def __init__(self, field, message):
        self.field = field
        self.message = message

    def json(self):
        j = dict()
        if self.field:
            j['field'] = self.field
        if self.message:
            j['reason'] = self.message
        return j

class Validation:
    def __init__(self, request):
        self.files = dict()
        self.errors = []
        self.status = 400
        if isinstance(request, dict):
            self.source = request
        else:
            contentType = request.headers.get("Content-Type")
            if contentType and contentType == "application/json":
                try:
                    self.source = json.loads(request.data.decode('utf-8'))
                    if not isinstance(self.source, dict):
                        self.error("Expected JSON dictionary")
                        self.source = {}
                except json.JSONDecodeError:
                    self.error("Invalid JSON provided")
                    self.source = {}
            else:
                self.source = request.form
                self.files = request.files
            self.request = request
        self._kwargs = {
            "valid": self,
            **self.source,
        }

    @property
    def ok(self):
        return len(self.errors) == 0

    def cls(self, name):
        return 'is-invalid' if any([
            e for e in self.errors if e.field == name
        ]) else ""

    def summary(self, name=None):
        errors = [e.message for e in self.errors if e.field == name or name == '@all']
        if len(errors) == 0:
            return ''
        if name is None:
            return Markup('<div class="alert alert-danger">{}</div>'
                    .format('<br />'.join(errors)))
        else:
            return Markup('<div class="invalid-feedback">{}</div>'
                    .format('<br />'.join(errors)))

    @property
    def response(self):
        return { "errors": [ e.json() for e in self.errors ] }, self.status

    @property
    def kwargs(self):
        return self._kwargs

    def error(self, message, field=None, status=None):
        self.errors.append(ValidationError(field, message))
        if status:
            self.status = status
        return self.response

    def optional(self, name, default=None, cls=None, max_file_size=-1):
        value = self.source.get(name)
        if value is None:
            value = self.files.get(name)
            if value and value.filename:
                if max_file_size >= 0:
                    fbytes = value.read(max_file_size + 1)
                    if len(fbytes) == max_file_size + 1:
                        self.error('{} is too large'.format(name), field=name)
                        return None
                    else:
                        value = fbytes
                else:
                    value = value.read()
        if value is None:
            if name in self.source:
                self.error('{} may not be null'.format(name), field=name)
                return None
            else:
                value = default
        if cls and value is not None:
            if cls and issubclass(cls, IntEnum):
                if not isinstance(value, int):
                    self.error('{} should be an int'.format(name), field=name)
                    return None
                else:
                    try:
                        value = cls(value)
                    except ValueError:
                        self.error('{} is not a valid {}'.format(
                            value, cls.__name__), field=name)
                        return None
            elif issubclass(cls, Enum):
                if not isinstance(value, str):
                    self.error("{} should be an str".format(name), field=name)
                else:
                    if value not in cls.__members__.keys():
                        self.error("{} should be a valid {}".format(name, cls.__name__),
                                field=name)
                        return None
                    else:
                        try:
                            value = cls[value]
                        except ValueError:
                            self.error('{} is not a valid {}'.format(
                                value, cls.__name__), field=name)
            elif not isinstance(value, cls):
                self.error('{} should be a {}'.format(name, cls.__name__), field=name)
                return None
        return value

    def require(self, name, cls=None, friendly_name=None):
        value = self.optional(name, None, cls)
        if not friendly_name:
            friendly_name = name
        if not isinstance(value, bool) and not value:
            self.error('{} is required'.format(friendly_name), field=name)
        return value

    def expect(self, condition, message, field=None, **kwargs):
        if not condition:
            self.error(message, field, **kwargs)

    def copy(self, valid, field=None):
        for err in self.errors:
            valid.error(err.message, field + "." + err.field)

    def error_for(self, *fields):
        for error in self.errors:
            if error.field in fields:
                return True
        return False

    def __contains__(self, value):
        return value in self.source or value in self.files

def valid_url(url):
    allowed_schemes = ('http', 'https', 'gemini', 'gopher')
    try:
        u = parse.urlparse(url)
        return bool(u.scheme and u.netloc and u.scheme in allowed_schemes)
    except:
        return False