-
Notifications
You must be signed in to change notification settings - Fork 0
/
api.py
274 lines (241 loc) · 8.43 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
from functools import wraps
from pathlib import Path
import os, requests
from flask import Flask, request, jsonify, Response
from flask_sqlalchemy import SQLAlchemy
from flask_limiter import Limiter, HEADERS
from flask_limiter.util import get_remote_address
AUTH_SERVER = 'http://10.0.20.21:9000'
app = Flask(__name__)
dir_path = os.path.dirname(os.path.realpath(__file__))
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///{}/teams.db'.format(dir_path)
db = SQLAlchemy(app)
def no_token():
return jsonify({'error' : 'no token in cookies'})
def no_match():
return jsonify({'error' : 'team number does not match token'})
"""
Query the auth server with passed cookie token
and returns the token's team number.
"""
def authentication(token):
data = {'token': token}
url = '{}/validate-session'.format(AUTH_SERVER)
resp = requests.post(url, data=data)
resp_data = resp.json()
if resp_data['success']:
return int(resp_data['success'])
else:
return jsonify({'error' : 'success key'})
"""
Compares the number returned from the auth server
with the teamnumber passed in the route.
"""
def blue_white(f):
@wraps(f)
def decorated(*args, **kwargs):
if 'token' not in request.cookies:
return no_token()
try:
token_num = authentication(request.cookies['token'])
except Exception:
return jsonify({'error' : 'success key not found'})
if token_num == 1337 or token_num == kwargs['teamNum']:
return f(*args, **kwargs)
else:
return no_match()
return decorated
"""
Compares the number returned from the auth server
with the teamnumber passed in the route.
"""
def white_team(f):
@wraps(f)
def decorated(*args, **kwargs):
if 'token' not in request.cookies:
return no_token()
try:
token_num = authentication(request.cookies['token'])
except Exception:
return jsonify({'error' : 'success key not found'})
if token_num == 1337:
return f(*args, **kwargs)
else:
return no_match()
return decorated
class Teams(db.Model):
teamNum = db.Column(db.Integer, unique=True, primary_key=True)
name = db.Column(db.String(50), unique=True)
guardian = db.Column(db.Integer, default=0)
bomber = db.Column(db.Integer, default=0)
striker = db.Column(db.Integer, default=0)
health = db.Column(db.Integer, default=100)
damage = db.Column(db.Integer, default=100)
speed = db.Column(db.Integer, default=100)
class Config(db.Model):
ratelimit = db.Column(db.Integer, primary_key=True)
def __init__(self, ratelimit=5):
self.ratelimit = ratelimit
db.session.commit()
"""
Creates a new team in the database with zeroed ships
"""
@app.route('/createteam', methods=['POST'])
@white_team
def create_team():
data = request.get_json(force=True)
new_team = Teams(teamNum=data['teamNum'], name=data['name'])
db.session.add(new_team)
db.session.commit()
return jsonify({'message' : 'Team {} - {} created'.format(data['teamNum'], data['name'])})
"""
Deletes an existing team in the database
"""
@app.route('/deleteteam/<int:teamNum>', methods=['DELETE'])
@white_team
def delete_team(teamNum):
team = Teams.query.filter_by(teamNum=teamNum).first()
if not team:
return jsonify({'message' : 'Team {} not found'.format(teamNum)})
db.session.delete(team)
db.session.commit()
return jsonify({'message' : 'Team {} deleted'.format(teamNum)})
"""
Returns JSON of all the teams in the database
"""
@app.route('/teams', methods=['GET'])
@white_team
def get_all_teams():
teams = Teams.query.all()
output = []
for team in teams:
team_data = {}
team_data['teamNum'] = team.teamNum
team_data['name'] = team.name
team_data['guardian'] = team.guardian
team_data['bomber'] = team.bomber
team_data['striker'] = team.striker
team_data['damage'] = team.damage
team_data['speed'] = team.speed
team_data['health'] = team.health
output.append(team_data)
return jsonify({'teams': output})
"""
Returns JSON of the team requested
"""
@app.route('/teams/<int:teamNum>', methods=['GET'])
@blue_white
def get_one_team(teamNum):
team = Teams.query.filter_by(teamNum=teamNum).first()
if not team:
return jsonify({'error' : 'Team {} - {} not found'.format(data['teamNum'], data['name'])})
team_data = {}
team_data['teamNum'] = team.teamNum
team_data['name'] = team.name
team_data['guardian'] = team.guardian
team_data['bomber'] = team.bomber
team_data['striker'] = team.striker
team_data['damage'] = team.damage
team_data['speed'] = team.speed
team_data['health'] = team.health
return jsonify(team_data)
"""
Override
"""
@app.route('/teams/<int:teamNum>', methods=['POST'])
@white_team
def override_one_team(teamNum):
team = Teams.query.filter_by(teamNum=teamNum).first()
if not team:
return jsonify({'error' : 'Team {} not found'.format(teamNum)})
data = request.get_json()
team.guardian = data['guardian']
team.bomber = data['bomber']
team.striker = data['striker']
team.damage = data['damage']
team.speed = data['speed']
team.health = data['health']
db.session.commit()
return jsonify({'message' : 'Team {} updated to guardian {}, bomber {}, striker {}, damage {}, speed {}, health {}'.format(teamNum, data['guardian'], data['bomber'], data['striker'], data['damage'], data['speed'], data['health'])})
"""
Resets a team's ship counts to default
"""
@app.route('/teams/<int:teamNum>/reset', methods=['POST'])
@white_team
def reset_one_team(teamNum):
team = Teams.query.filter_by(teamNum=teamNum).first()
if not team:
return jsonify({'error' : 'Team {} not found'.format(teamNum)})
team.guardian = 0
team.bomber = 0
team.striker = 0
team.damage = 100
team.speed = 100
team.health = 100
db.session.commit()
return jsonify({'message' : 'Team {} has been reset'.format(teamNum)})
"""
Increments a team's guardian ship count
"""
@app.route('/teams/<int:teamNum>/guardian', methods=['POST'])
@white_team
def increment_guardian(teamNum):
team = Teams.query.filter_by(teamNum=teamNum).first()
if not team:
return jsonify({'error' : 'Team {} not found'.format(teamNum)})
data = request.get_json(force=True)
team.guardian += data['value']
if team.guardian < 0:
team.guardian = 0
db.session.commit()
return jsonify({'message' : 'Team {} has built {} guardian ships'.format(teamNum, data['value'])})
"""
Increments a team's bomber ship count
"""
@app.route('/teams/<int:teamNum>/bomber', methods=['POST'])
@white_team
def increment_bomber(teamNum):
team = Teams.query.filter_by(teamNum=teamNum).first()
if not team:
return jsonify({'error' : 'Team {} not found'.format(teamNum)})
data = request.get_json(force=True)
team.bomber += data['value']
if team.bomber < 0:
team.bomber = 0
db.session.commit()
return jsonify({'message' : 'Team {} has built {} bomber ships'.format(teamNum, data['value'])})
"""
Increments a team's striker ship count
"""
@app.route('/teams/<int:teamNum>/striker', methods=['POST'])
@blue_white
def increment_striker(teamNum):
team = Teams.query.filter_by(teamNum=teamNum).first()
if not team:
return jsonify({'error' : 'Team {} not found'.format(teamNum)})
data = request.get_json(force=True)
team.striker += data['value']
if team.striker < 0:
team.striker = 0
db.session.commit()
return jsonify({'message' : 'Team {} has built {} striker ships'.format(teamNum, data['value'])})
"""
Boost damage, health, or speed by however much was passed.
"""
@app.route('/teams/<int:teamNum>/boost', methods=['POST'])
@blue_white
def boost_team(teamNum):
team = Teams.query.filter_by(teamNum=teamNum).first()
if not team:
return jsonify({'error' : 'Team {} not found'.format(teamNum)})
data = request.get_json(force=True)
try:
setattr(team, data['type'], getattr(team, data['type'])+data['value'])
except Exception as e:
return jsonify({'error' : '{}'.format(e)})
db.session.commit()
return jsonify({'message' : 'Team {} {}\'d their {} by {} '.format(teamNum, data['change'], data['type'], data['value'])})
if __name__ == '__main__':
if not os.path.exists('teams.db'):
open('teams.db', 'w').close()
app.run(debug=True, host='0.0.0.0', port=6000)