-
Notifications
You must be signed in to change notification settings - Fork 41
/
api.py
303 lines (269 loc) · 9.61 KB
/
api.py
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
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
# Standard library imports...
import logging
from requests import exceptions
import re
from urllib.parse import urlparse
import os
# Third-party imports...
from datetime import datetime
from flask import (
Blueprint,
abort,
jsonify,
request,
make_response,
render_template
)
from flask_httpauth import HTTPTokenAuth
from wtforms import (
Form,
StringField,
PasswordField,
validators
)
# Local imports...
from .matrix_api import create_account
from . import config
from . import tokens
from .constants import __location__
auth = HTTPTokenAuth(scheme='SharedSecret')
logger = logging.getLogger(__name__)
api = Blueprint("api", __name__)
re_mxid = re.compile(r'^@?[a-zA-Z_\-=\.\/0-9]+(:[a-zA-Z\-\.:\/0-9]+)?$')
def validate_token(form, token):
"""
validates token
Parameters
----------
arg1 : Form object
arg2 : str
token name, e.g. 'DoubleWizardSki'
Raises
-------
ValidationError
Token is invalid
"""
tokens.tokens.load()
if not tokens.tokens.active(token.data):
raise validators.ValidationError('Token is invalid')
def validate_username(form, username):
"""
validates username
Parameters
----------
arg1 : Form object
arg2 : str
username name, e.g: '@user:matrix.org' or 'user'
https://github.com/matrix-org/matrix-doc/blob/master/specification/appendices/identifier_grammar.rst#user-identifiers
Raises
-------
ValidationError
Username doesn't follow mxid requirements
"""
domain = urlparse(config.config.server_location).hostname
re_mxid = r'^@?[a-zA-Z_\-=\.\/0-9]+(:' + \
re.escape(domain) + \
r')?$'
err = "Username doesn't follow pattern: '%s'" % re_mxid
if not re.search(re_mxid, username.data):
raise validators.ValidationError(err)
def validate_password(form, password):
"""
validates username
Parameters
----------
arg1 : Form object
arg2 : str
password
Raises
-------
ValidationError
Password doesn't follow length requirements
"""
min_length = config.config.password['min_length']
err = 'Password should be between %s and 255 chars long' % min_length
if len(password.data) < min_length or len(password.data) > 255:
raise validators.ValidationError(err)
class RegistrationForm(Form):
"""
Registration Form
validates user account registration requests
"""
username = StringField('Username', [
validators.Length(min=1, max=200),
# validators.Regexp(re_mxid)
validate_username
])
password = PasswordField('New Password', [
# validators.Length(min=8),
validate_password,
validators.DataRequired(),
validators.EqualTo('confirm', message='Passwords must match')
])
confirm = PasswordField('Repeat Password')
token = StringField('Token', [
validators.Regexp(r'^([A-Z][a-z]+)+$'),
validate_token
])
@auth.verify_token
def verify_token(token):
return token != 'APIAdminPassword' and token == config.config.admin_secret
@auth.error_handler
def unauthorized():
resp = {
'errcode': 'MR_BAD_SECRET',
'error': 'wrong shared secret'
}
return make_response(jsonify(resp), 401)
@api.route('/register', methods=['GET', 'POST'])
def register():
"""
main user account registration endpoint
to register an account you need to send a
application/x-www-form-urlencoded request with
- username
- password
- confirm
- token
as described in the RegistrationForm
"""
if request.method == 'POST':
logger.debug('an account registration started...')
form = RegistrationForm(request.form)
logger.debug('validating request data...')
if form.validate():
logger.debug('request valid')
# remove sigil and the domain from the username
username = form.username.data.rsplit(':')[0].split('@')[-1]
logger.debug('creating account %s...' % username)
# send account creation request to the hs
try:
account_data = create_account(form.username.data,
form.password.data,
config.config.server_location,
config.config.shared_secret)
except exceptions.ConnectionError:
logger.error('can not connect to %s' % config.config.server_location,
exc_info=True)
abort(500)
except exceptions.HTTPError as e:
resp = e.response
error = resp.json()
status_code = resp.status_code
if status_code == 404:
logger.error('no HS found at %s' %
config.config.server_location)
elif status_code == 403:
logger.error(
'wrong shared registration secret or not enabled')
elif status_code == 400:
# most likely this should only be triggered if a userid
# is already in use
return make_response(jsonify(error), 400)
else:
logger.error('failure communicating with HS',
exc_info=True)
abort(500)
logger.debug('using token %s' % form.token.data)
ip = request.remote_addr if config.config.ip_logging else False
tokens.tokens.use(form.token.data, ip)
logger.debug('account creation succeded!')
return jsonify(access_token=account_data['access_token'],
home_server=account_data['home_server'],
user_id=account_data['user_id'],
status='success',
status_code=200)
else:
logger.debug('account creation failed!')
resp = {'errcode': 'MR_BAD_USER_REQUEST',
'error': form.errors}
return make_response(jsonify(resp), 400)
# for fieldName, errorMessages in form.errors.items():
# for err in errorMessages:
# # return error to user
else:
server_name = config.config.server_name
pw_length = config.config.password['min_length']
return render_template('register.html',
server_name=server_name,
pw_length=pw_length,
riot_instance=config.config.riot_instance,
base_url=config.config.base_url)
@api.route('/api/version')
@auth.login_required
def version():
with open(os.path.join(__location__, '__init__.py'), 'r') as file:
version_file = file.read()
version_match = re.search(r"^__version__ = ['\"]([^'\"]*)['\"]",
version_file, re.M)
resp = {
'version': version_match.group(1),
}
return make_response(jsonify(resp), 200)
@api.route('/api/token', methods=['GET', 'POST'])
@auth.login_required
def token():
tokens.tokens.load()
data = False
max_usage = False
expiration_date = None
if request.method == 'GET':
return jsonify(tokens.tokens.toList())
elif request.method == 'POST':
data = request.get_json()
try:
if data:
if 'expiration_date' in data and data['expiration_date'] is not None:
expiration_date = datetime.fromisoformat(data['expiration_date'])
if 'max_usage' in data:
max_usage = data['max_usage']
token = tokens.tokens.new(expiration_date=expiration_date,
max_usage=max_usage)
except ValueError:
resp = {
'errcode': 'MR_BAD_DATE_FORMAT',
'error': "date wasn't in YYYY-MM-DD format"
}
return make_response(jsonify(resp), 400)
return jsonify(token.toDict())
resp = {
'errcode': 'MR_BAD_USER_REQUEST',
'error': 'malformed request'
}
return make_response(jsonify(resp), 400)
@api.route('/api/token/<token>', methods=['GET', 'PATCH'])
@auth.login_required
def token_status(token):
tokens.tokens.load()
data = False
if request.method == 'GET':
if tokens.tokens.get_token(token):
return jsonify(tokens.tokens.get_token(token).toDict())
else:
resp = {
'errcode': 'MR_TOKEN_NOT_FOUND',
'error': 'token does not exist'
}
return make_response(jsonify(resp), 404)
elif request.method == 'PATCH':
data = request.get_json(force=True)
if data:
if 'ips' not in data and 'active' not in data and 'name' not in data:
if tokens.tokens.update(token, data):
return jsonify(tokens.tokens.get_token(token).toDict())
else:
resp = {
'errcode': 'MR_BAD_USER_REQUEST',
'error': 'you\'re not allowed to change this property'
}
return make_response(jsonify(resp), 400)
resp = {
'errcode': 'MR_TOKEN_NOT_FOUND',
'error': 'token does not exist'
}
return make_response(jsonify(resp), 404)
resp = {
'errcode': 'MR_BAD_USER_REQUEST',
'error': 'malformed request'
}
return make_response(jsonify(resp), 400)