aboutsummaryrefslogtreecommitdiff
path: root/app/blueprints/api/tokens.py
blob: 5ef870dd72f490754b213651019b9b3143b399ae (plain)
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
# ContentDB
# Copyright (C) 2018  rubenwardy
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <https://www.gnu.org/licenses/>.


from flask import render_template, redirect, request, session, url_for, abort
from flask_user import login_required, current_user
from . import bp
from app.models import db, User, APIToken, Package, Permission
from app.utils import randomString
from app.querybuilder import QueryBuilder

from flask_wtf import FlaskForm
from wtforms import *
from wtforms.validators import *
from wtforms.ext.sqlalchemy.fields import QuerySelectField

class CreateAPIToken(FlaskForm):
	name	     = StringField("Name", [InputRequired(), Length(1, 30)])
	package      = QuerySelectField("Limit to package", allow_blank=True, \
			get_pk=lambda a: a.id, get_label=lambda a: a.title)
	submit	     = SubmitField("Save")


@bp.route("/user/tokens/")
@login_required
def list_tokens_redirect():
	return redirect(url_for("api.list_tokens", username=current_user.username))


@bp.route("/users/<username>/tokens/")
@login_required
def list_tokens(username):
	user = User.query.filter_by(username=username).first()
	if user is None:
		abort(404)

	if not user.checkPerm(current_user, Permission.CREATE_TOKEN):
		abort(403)

	return render_template("api/list_tokens.html", user=user)


@bp.route("/users/<username>/tokens/new/", methods=["GET", "POST"])
@bp.route("/users/<username>/tokens/<int:id>/edit/", methods=["GET", "POST"])
@login_required
def create_edit_token(username, id=None):
	user = User.query.filter_by(username=username).first()
	if user is None:
		abort(404)

	if not user.checkPerm(current_user, Permission.CREATE_TOKEN):
		abort(403)

	is_new = id is None

	token = None
	access_token = None
	if not is_new:
		token = APIToken.query.get(id)
		if token is None:
			abort(404)
		elif token.owner != user:
			abort(403)

		access_token = session.pop("token_" + str(token.id), None)

	form = CreateAPIToken(formdata=request.form, obj=token)
	form.package.query_factory = lambda: Package.query.filter_by(author=user).all()

	if request.method == "POST" and form.validate():
		if is_new:
			token = APIToken()
			token.owner = user
			token.access_token = randomString(32)

		form.populate_obj(token)
		db.session.add(token)
		db.session.commit() # save

		if is_new:
			# Store token so it can be shown in the edit page
			session["token_" + str(token.id)] = token.access_token

		return redirect(url_for("api.create_edit_token", username=username, id=token.id))

	return render_template("api/create_edit_token.html", user=user, form=form, token=token, access_token=access_token)


@bp.route("/users/<username>/tokens/<int:id>/reset/", methods=["POST"])
@login_required
def reset_token(username, id):
	user = User.query.filter_by(username=username).first()
	if user is None:
		abort(404)

	if not user.checkPerm(current_user, Permission.CREATE_TOKEN):
		abort(403)

	token = APIToken.query.get(id)
	if token is None:
		abort(404)
	elif token.owner != user:
		abort(403)

	token.access_token = randomString(32)

	db.session.commit() # save

	# Store token so it can be shown in the edit page
	session["token_" + str(token.id)] = token.access_token

	return redirect(url_for("api.create_edit_token", username=username, id=token.id))


@bp.route("/users/<username>/tokens/<int:id>/delete/", methods=["POST"])
@login_required
def delete_token(username, id):
	user = User.query.filter_by(username=username).first()
	if user is None:
		abort(404)

	if not user.checkPerm(current_user, Permission.CREATE_TOKEN):
		abort(403)

	is_new = id is None

	token = APIToken.query.get(id)
	if token is None:
		abort(404)
	elif token.owner != user:
		abort(403)

	db.session.delete(token)
	db.session.commit()

	return redirect(url_for("api.list_tokens", username=username))