-
Notifications
You must be signed in to change notification settings - Fork 91
Expand file tree
/
Copy pathvalidator.py
More file actions
141 lines (112 loc) · 3.62 KB
/
validator.py
File metadata and controls
141 lines (112 loc) · 3.62 KB
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
# Patchwork - automated patch tracking system
# Copyright (C) 2018 Stephen Finucane <stephen@that.guru>
#
# SPDX-License-Identifier: GPL-2.0-or-later
import os
import re
from django.urls import resolve
import jsonschema_path
from openapi_core.contrib.django import DjangoOpenAPIRequest
from openapi_core.contrib.django import DjangoOpenAPIResponse
from openapi_core.exceptions import OpenAPIError
from openapi_core.templating import util
from openapi_core.validation.request.exceptions import SecurityValidationError
from openapi_core import shortcuts
from rest_framework import status
import yaml
# docs/api/schemas
SCHEMAS_DIR = os.path.join(
os.path.dirname(os.path.abspath(__file__)),
os.pardir,
os.pardir,
os.pardir,
'docs',
'api',
'schemas',
)
_LOADED_SPECS = {}
# HACK! Workaround for https://github.com/p1c2u/openapi-core/issues/226
def search(path_pattern, full_url_pattern):
p = util.Parser(path_pattern)
p._expression = p._expression + '$'
result = p.search(full_url_pattern)
if not result or any('/' in arg for arg in result.named.values()):
return None
return result
util.search = search
class RegexValidator(object):
def __init__(self, regex):
self.regex = re.compile(regex, re.IGNORECASE)
def __call__(self, value):
if not isinstance(value, str):
return False
if not value:
return True
return self.regex.match(value)
EXTRA_FORMAT_VALIDATORS = {
'uri': RegexValidator(
r'^(?:http|ftp)s?://'
r'(?:(?:[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?\.)+(?:[A-Z]{2,6}\.?|[A-Z0-9-]{2,}\.?)|' # noqa: E501
r'localhost|'
r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})'
r'(?::\d+)?'
r'(?:/?|[/?]\S+)$',
),
'iso8601': RegexValidator(r'^\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d\.\d{6}$'),
'email': RegexValidator(r'[^@]+@[^@]+\.[^@]+'),
}
def _load_spec(version):
global _LOADED_SPECS
if _LOADED_SPECS.get(version):
return _LOADED_SPECS[version]
spec_path = os.path.join(
SCHEMAS_DIR,
'v{}'.format(version) if version else 'latest',
'patchwork.yaml',
)
with open(spec_path, 'r') as fh:
data = yaml.load(fh, Loader=yaml.SafeLoader)
_LOADED_SPECS[version] = jsonschema_path.SchemaPath.from_dict(data)
return _LOADED_SPECS[version]
def validate_data(
path,
request,
response,
validate_request,
validate_response,
):
if response.status_code in (
# status.HTTP_403_FORBIDDEN,
status.HTTP_405_METHOD_NOT_ALLOWED,
):
return
# FIXME: this shouldn't matter
request.path = request.path.rstrip('/')
spec = _load_spec(resolve(path).kwargs.get('version'))
request = DjangoOpenAPIRequest(request)
response = DjangoOpenAPIResponse(response)
# request
if validate_request:
try:
shortcuts.validate_request(
request,
spec=spec,
extra_format_validators=EXTRA_FORMAT_VALIDATORS,
)
except SecurityValidationError:
assert response.status_code in (
status.HTTP_403_FORBIDDEN,
status.HTTP_404_NOT_FOUND,
)
except OpenAPIError:
# TODO(stephenfin): In API v2.0, this should be an error. As things
# stand, we silently ignore these issues.
assert response.status_code == status.HTTP_200_OK
# response
if validate_response:
shortcuts.validate_response(
request,
response,
spec=spec,
extra_format_validators=EXTRA_FORMAT_VALIDATORS,
)