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
|
import base64
import re
import traceback
import uuid
import bcrypt
import flask
from sqlalchemy.exc import IntegrityError
from sqlalchemy.orm.exc import NoResultFound
from auth import Auth, InvalidTokenError
from database import User, DbSession, ValidationError
import mapping
from myconfig import mappings
app = flask.Flask(__name__)
def cb_decorator(fn):
return lambda *args, **kwargs: lambda cb: fn(*args, **kwargs, cb = cb)
@cb_decorator
def db_route(path, cb):
@app.route(path, endpoint = path)
def db_handler(*args, **kwargs):
dbsession = DbSession()
try:
ret = cb(*args, **kwargs, dbsession = dbsession)
except (ValidationError, NoResultFound):
traceback.print_exc()
return flask.Response(status = 400)
try:
dbsession.commit()
except IntegrityError as ex:
# 1062 is the SQL error code for duplicate entry.
# That's a user input error, but if we get some other kind
# of integrity error, then it's a server error.
if ex.orig.args[0] == 1062:
return flask.Response(status = 400)
else:
raise
dbsession.close()
return ret
return db_handler
# /Auth/ endpoints use basic authentication and create new tokens.
@cb_decorator
def auth_route(path, cb):
@db_route(f"/Auth{path}")
def auth_handler(*args, dbsession, **kwargs):
try:
kind, value = flask.request.headers["Authentication"].split(" ")
assert kind == "Basic"
email, password = base64.b64decode(value).decode().split(":")
except (KeyError, ValueError, AssertionError):
print("no basic auth")
return flask.Response(status = 401)
email = email.strip()
return cb(*args, **kwargs, dbsession = dbsession, email = email, password = password)
return auth_handler
# /API/ endpoints use token authentication and operate as a
# Unite user.
@cb_decorator
def api_route(path, cb):
@db_route(f"/API{path}")
def api_handler(*args, dbsession, **kwargs):
try:
kind, token = flask.request.headers["Authentication"].split(" ")
assert kind == "Token"
auth = Auth(dbsession, flask.request.remote_addr)
authsession = auth.session(token)
except (KeyError, ValueError, AssertionError, InvalidTokenError):
return flask.Response(status = 401)
return cb(*args, **kwargs, dbsession = dbsession, authsession = authsession)
return api_handler
@auth_route("/Register")
def register(email, password, dbsession):
def error(message):
return flask.jsonify({"ClassName":"System.Exception","Message":message,"Data":None,"InnerException":None,"HelpURL":None,"StackTraceString":None,"RemoteStackTraceString":None,"RemoteStackIndex":0,"ExceptionMethod":None,"HResult":-2146233088,"Source":None})
try:
displayname = flask.request.args["displayname"]
sysname = flask.request.args["sysname"]
appname = flask.request.args["appname"][:255]
appdesc = flask.request.args["appdesc"][:255]
version = flask.request.args["version"][:255]
except KeyError as ex:
return error(str(ex))
# Additional constraints for password
if len(password) < 7:
return error("Password too short")
requirements = [".*[A-Z].*", ".*[a-z].*", ".*[0-9].*"]
for expr in list(requirements):
if re.compile(expr).match(password) is None:
return error("Password does not meet the requirements")
password = bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()
try:
user = User(ID = str(uuid.uuid4()), Email = email, Password = password, DisplayName = displayname, SysName = sysname)
except ValidationError:
return error(str(ex))
dbsession.add(user)
try:
dbsession.commit()
except IntegrityError as ex:
if ex.orig.args[0] == 1062:
dbsession.rollback()
return error(ex.orig.args[1])
else:
raise
auth = Auth(dbsession, flask.request.remote_addr)
return auth.create_session(user, appname, appdesc, version).Token
@auth_route("/Login")
def login(email, password, dbsession):
if email == "" or password == "":
return flask.Response(status = 400)
try:
appname = flask.request.args["appname"][:255]
appdesc = flask.request.args["appdesc"][:255]
version = flask.request.args["version"][:255]
except KeyError:
return flask.Response(status = 400)
try:
user = dbsession.query(User).filter_by(Email = email).one()
except NoResultFound:
print("acct does not exist")
return flask.Response(status = 401)
if user.Password is None: # login disabled
print("login disabled")
return flask.Response(status = 401)
if not bcrypt.checkpw(password.encode(), user.Password.encode()):
print("password incorrect")
return flask.Response(status = 401)
auth = Auth(dbsession, flask.request.remote_addr)
return auth.create_session(user, appname, appdesc, version).Token
@api_route("/GetDisplayName/<uid>")
def get_display_name(uid, dbsession, authsession):
return dbsession.query(User).filter_by(ID = uid).one().DisplayName
@api_route("/GetPongHighscores")
def get_pong_highscores(dbsession, authsession):
return flask.jsonify({"Pages": 0, # unused
"Highscores": [mapping.map_to(user, mappings["PongHighscore"])
for user in dbsession.query(User).filter(User.PongLevel.isnot(None)).filter(User.PongCP.isnot(None))
.order_by(User.PongCP.desc()).all()]})
@api_route("/GetEmail")
def get_email(dbsession, authsession):
return authsession.User.Email
def gettersetters(name, convert = lambda v: v):
@api_route(f"/Get{name}")
def getter(dbsession, authsession):
return str(getattr(authsession.User, name))
@api_route(f"/Set{name}/<value>")
def setter(value, dbsession, authsession):
setattr(authsession.User, name, convert(value))
return flask.Response(status = 200)
@api_route("/GetPongCP")
def get_pong_cp(dbsession, authsession):
return str(authsession.User.PongCP or 0)
@api_route("/SetPongCP/<value>")
def set_pong_cp(value, dbsession, authsession):
value = int(value)
if authsession.User.PongCP is not None and value <= authsession.User.PongCP:
return flask.Response(status = 400)
authsession.User.PongCP = value
return flask.Response(status = 200)
@api_route("/GetPongLevel")
def get_pong_level(dbsession, authsession):
return str(authsession.User.PongLevel or 0)
@api_route("/SetPongLevel/<value>")
def set_pong_cp(value, dbsession, authsession):
value = int(value)
if authsession.User.PongLevel is not None and value <= authsession.User.PongLevel:
return flask.Response(status = 400)
authsession.User.PongLevel = value
return flask.Response(status = 200)
gettersetters("SysName")
gettersetters("DisplayName")
gettersetters("FullName")
gettersetters("Codepoints", convert = int)
|
