easyctf-2017/server/api/user.py

333 lines
10 KiB
Python
Raw Normal View History

2016-03-03 05:36:51 +00:00
from flask import Blueprint, make_response, session, request, redirect, url_for, send_file
2016-03-04 00:27:10 +00:00
from werkzeug import secure_filename
2015-12-24 02:06:49 +00:00
from flask import current_app as app
2016-01-06 06:15:57 +00:00
from voluptuous import Schema, Length, Required
2015-12-24 02:06:49 +00:00
2016-01-06 06:15:57 +00:00
from models import db, LoginTokens, Users
from decorators import api_wrapper, WebException
from schemas import verify_to_schema, check
2015-12-23 23:23:18 +00:00
2016-01-07 00:23:43 +00:00
import datetime
2015-12-27 01:21:15 +00:00
import logger
2016-03-04 00:27:10 +00:00
import os
2016-01-06 06:15:57 +00:00
import re
2015-12-24 04:31:50 +00:00
import requests
2016-01-16 06:53:35 +00:00
import team
2015-12-24 02:30:51 +00:00
import utils
2016-03-04 00:27:10 +00:00
from PIL import Image
2016-01-06 06:15:57 +00:00
###############
# USER ROUTES #
###############
2015-12-23 23:23:18 +00:00
blueprint = Blueprint("user", __name__)
2016-01-16 22:36:30 +00:00
@blueprint.route("/forgot", methods=["POST"])
@blueprint.route("/forgot/<token>", methods=["GET", "POST"])
@api_wrapper
def user_forgot_password(token=None):
2016-01-18 06:41:11 +00:00
params = utils.flat_multi(request.form)
if token is not None:
user = get_user(reset_token=token).first()
if user is None:
raise WebException("Invalid reset token.")
# We are viewing the actual reset form
if request.method == "GET":
return { "success": 1, "message": ""}
# Submission of actual reset form
if request.method == "POST":
password = params.get("password")
confirm_password = params.get("confirm_password")
if password != confirm_password:
raise WebException("Passwords do not match.")
else:
user.password = utils.hash_password(password)
user.reset_token = None
current_session = db.session.object_session(user)
current_session.add(user)
current_session.commit()
return { "success": 1, "message": "Success!" }
else:
email = params.get("email").lower()
user = get_user(email=email).first()
if user is None:
raise WebException("User with that email does not exist.")
token = utils.generate_string(length=64)
user.reset_token = token
current_session = db.session.object_session(user)
current_session.add(user)
current_session.commit()
reset_link = "%s/forgot/%s" % ("127.0.0.1:8000", token)
subject = "EasyCTF password reset"
body = """%s,\n\nA request to reset your EasyCTF password has been made. If you did not request this password reset, you may safely ignore this email and delete it.\n\nYou may reset your password by clicking this link or pasting it to your browser.\n\n%s\n\nThis link can only be used once, and will lead you to a page where you can reset your password.\n\nGood luck!\n\n- The EasyCTF Team""" % (user.username, reset_link)
response = utils.send_email(email, subject, body).json()
if "Queued" in response["message"]:
return { "success": 1, "message": "Email sent to %s" % email }
else:
raise WebException(response["message"])
2016-01-16 22:36:30 +00:00
2015-12-23 23:23:18 +00:00
@blueprint.route("/register", methods=["POST"])
@api_wrapper
def user_register():
2016-01-06 06:15:57 +00:00
params = utils.flat_multi(request.form)
name = params.get("name")
email = params.get("email")
username = params.get("username")
password = params.get("password")
password_confirm = params.get("password_confirm")
utype = int(params.get("type"))
if password != password_confirm:
raise WebException("Passwords do not match.")
verify_to_schema(UserSchema, params)
user = Users(name, username, email, password, utype)
with app.app_context():
db.session.add(user)
db.session.commit()
2016-03-03 05:36:51 +00:00
utils.generate_identicon(email, user.uid)
2016-01-06 06:15:57 +00:00
2016-01-18 06:41:11 +00:00
logger.log(__name__, logger.INFO, "%s registered with %s" % (name.encode("utf-8"), email.encode("utf-8")))
2016-01-06 06:35:59 +00:00
login_user(username, password)
2016-01-06 06:15:57 +00:00
2016-01-06 06:35:59 +00:00
return { "success": 1, "message": "Success!" }
2016-01-06 06:15:57 +00:00
2016-01-08 03:25:50 +00:00
@blueprint.route("/logout", methods=["GET"])
2015-12-24 00:54:47 +00:00
@api_wrapper
def user_logout():
2016-01-06 06:15:57 +00:00
sid = session["sid"]
username = session["username"]
2016-01-06 06:35:59 +00:00
with app.app_context():
expired = LoginTokens.query.filter_by(username=username).all()
for expired_token in expired: db.session.delete(expired_token)
db.session.commit()
2016-01-06 06:15:57 +00:00
session.clear()
2015-12-24 00:54:47 +00:00
@blueprint.route("/login", methods=["POST"])
@api_wrapper
def user_login():
2016-01-06 06:15:57 +00:00
params = utils.flat_multi(request.form)
username = params.get("username")
password = params.get("password")
result = login_user(username, password)
if result != True:
raise WebException("Please check if your username/password are correct.")
return { "success": 1, "message": "Success!" }
2015-12-24 02:06:49 +00:00
2016-01-08 03:25:50 +00:00
@blueprint.route("/status", methods=["GET"])
@api_wrapper
def user_status():
2016-01-06 06:15:57 +00:00
logged_in = is_logged_in()
result = {
"success": 1,
"logged_in": logged_in,
"admin": is_admin(),
"username": session["username"] if logged_in else "",
}
2016-01-12 03:54:26 +00:00
if logged_in:
result["has_team"] = in_team(get_user().first())
2016-01-08 03:25:50 +00:00
2016-01-06 06:15:57 +00:00
return result
2016-01-08 03:25:50 +00:00
@blueprint.route("/info", methods=["GET"])
2016-01-07 00:23:43 +00:00
@api_wrapper
def user_info():
logged_in = is_logged_in()
2016-01-12 03:54:26 +00:00
username = utils.flat_multi(request.args).get("username")
2016-01-07 00:23:43 +00:00
if username is None:
if logged_in:
username = session["username"]
if username is None:
raise WebException("No user specified.")
2016-01-07 06:23:00 +00:00
me = False if not("username" in session) else username.lower() == session["username"].lower()
2016-01-07 00:23:43 +00:00
user = get_user(username_lower=username.lower()).first()
if user is None:
2016-01-16 16:41:16 +00:00
raise WebException("User not found.")
2016-01-07 00:23:43 +00:00
show_email = me if logged_in else False
2016-01-16 06:53:35 +00:00
user_in_team = in_team(user)
2016-01-07 00:23:43 +00:00
userdata = {
"user_found": True,
"name": user.name,
"username": user.username,
"type": ["Student", "Instructor", "Observer"][user.utype - 1],
"admin": user.admin,
"registertime": datetime.datetime.fromtimestamp(user.registertime).isoformat() + "Z",
"me": me,
2016-01-16 06:53:35 +00:00
"show_email": show_email,
2016-03-03 05:36:51 +00:00
"in_team": user_in_team,
"uid": user.uid
2016-01-07 00:23:43 +00:00
}
if show_email:
userdata["email"] = user.email
2016-01-16 06:53:35 +00:00
if user_in_team:
userdata["team"] = team.get_team_info(tid=user.tid)
2016-02-29 18:35:40 +00:00
if me and not(user_in_team):
invitations = user.get_invitations()
userdata["invitations"] = invitations
2016-01-07 00:23:43 +00:00
return { "success": 1, "user": userdata }
2016-03-03 05:36:51 +00:00
@blueprint.route("/avatar/<uid>", methods=["GET"])
def user_avatar(uid):
uid = int(uid)
try:
return send_file("pfp/%d.png" % uid, mimetype="image/png")
except:
user = get_user(uid=uid).first()
if user is not None:
utils.generate_identicon(user.email, user.uid)
return send_file("pfp/%d.png" % uid, mimetype="image/png")
return abort(404)
2016-03-04 00:27:10 +00:00
@blueprint.route("/avatar/upload", methods=["POST"])
@api_wrapper
def user_avatar_upload():
logged_in = is_logged_in()
if not logged_in:
raise WebException("You're not logged in.")
_user = get_user().first()
f = request.files["file"]
if f is None:
raise WebException("Please upload something.")
fname = "/tmp/" + secure_filename(utils.generate_string())
f.save(fname)
try:
pfp = "pfp/%d.png" % _user.uid
os.remove(pfp)
im = Image.open(fname)
im = im.resize((256, 256), Image.ANTIALIAS)
im.save(open(pfp, "w"), "PNG")
return { "success": 1, "message": "Uploaded!" }
except Exception, e:
raise WebException(str(e))
@blueprint.route("/avatar/remove", methods=["POST"])
@api_wrapper
def user_avatar_remove():
logged_in = is_logged_in()
if not logged_in:
raise WebException("You're not logged in.")
_user = get_user().first()
try:
pfp = "pfp/%d.png" % _user.uid
os.remove(pfp)
return { "success": 1, "message": "Removed!" }
except Exception, e:
raise WebException(str(e))
2016-01-06 06:15:57 +00:00
##################
# USER FUNCTIONS #
##################
__check_username = lambda username: get_user(username_lower=username.lower()).first() is None
__check_email = lambda email: get_user(email=email.lower()).first() is None
UserSchema = Schema({
Required("email"): check(
([str, Length(min=4, max=128)], "Your email should be between 4 and 128 characters long."),
([__check_email], "Someone already registered this email."),
2016-01-12 03:54:26 +00:00
([utils.__check_email_format], "Please enter a legit email.")
2016-01-06 06:15:57 +00:00
),
Required("name"): check(
([str, Length(min=4, max=128)], "Your name should be between 4 and 128 characters long.")
),
Required("username"): check(
([str, Length(min=4, max=32)], "Your username should be between 4 and 32 characters long."),
2016-01-16 21:12:32 +00:00
([utils.__check_alphanumeric], "Please only use alphanumeric characters in your username."),
2016-01-06 06:15:57 +00:00
([__check_username], "This username is taken, did you forget your password?")
),
Required("password"): check(
([str, Length(min=4, max=64)], "Your password should be between 4 and 64 characters long."),
2016-01-12 03:54:26 +00:00
([utils.__check_ascii], "Please only use ASCII characters in your password."),
2016-01-06 06:15:57 +00:00
),
Required("type"): check(
([str, lambda x: x.isdigit()], "Please use the online form.")
),
"notify": str
}, extra=True)
2016-01-16 22:36:30 +00:00
def get_user(username=None, username_lower=None, email=None, uid=None, reset_token=None):
2016-01-18 06:41:11 +00:00
match = {}
if username != None:
match.update({ "username": username })
elif username_lower != None:
match.update({ "username_lower": username_lower })
elif uid != None:
match.update({ "uid": uid })
elif email != None:
match.update({ "email": email })
elif is_logged_in():
match.update({ "username": session["username"] })
elif reset_token != None:
match.update({ "reset_token": reset_token })
with app.app_context():
result = Users.query.filter_by(**match)
return result
2016-01-06 06:15:57 +00:00
def login_user(username, password):
user = get_user(username_lower=username.lower()).first()
if user is None: return False
correct = utils.check_password(user.password, password)
if not correct: return False
useragent = request.headers.get("User-Agent")
ip = request.remote_addr
2016-01-06 06:35:59 +00:00
2016-01-06 06:15:57 +00:00
with app.app_context():
2016-01-06 06:35:59 +00:00
expired = LoginTokens.query.filter_by(username=username).all()
for expired_token in expired: db.session.delete(expired_token)
token = LoginTokens(user.uid, user.username, ua=useragent, ip=ip)
2016-01-06 06:15:57 +00:00
db.session.add(token)
db.session.commit()
2016-01-16 16:41:16 +00:00
2016-01-06 06:15:57 +00:00
session["sid"] = token.sid
session["username"] = token.username
2016-01-07 00:23:43 +00:00
session["admin"] = user.admin == True
2016-01-18 06:41:11 +00:00
if user.tid is not None and user.tid >= 0:
session["tid"] = user.tid
2016-01-06 06:15:57 +00:00
2016-01-06 06:35:59 +00:00
return True
2016-01-12 03:54:26 +00:00
def in_team(user):
return user.tid is not None and user.tid >= 0
def is_logged_in():
2016-01-07 06:23:00 +00:00
if not("sid" in session and "username" in session): return False
2016-01-06 06:15:57 +00:00
sid = session["sid"]
username = session["username"]
token = LoginTokens.query.filter_by(sid=sid).first()
if token is None: return False
2016-01-06 06:15:57 +00:00
useragent = request.headers.get("User-Agent")
ip = request.remote_addr
if token.username != username: return False
if token.ua != useragent: return False
return True
2016-01-06 06:15:57 +00:00
def is_admin():
return is_logged_in() and "admin" in session and session["admin"]
2015-12-24 04:31:50 +00:00
def validate_captcha(form):
2016-01-06 06:15:57 +00:00
if "captcha_response" not in form:
return False
captcha_response = form["captcha_response"]
data = {"secret": "6Lc4xhMTAAAAACFaG2NyuKoMdZQtSa_1LI76BCEu", "response": captcha_response}
response = requests.post("https://www.google.com/recaptcha/api/siteverify", data=data)
2016-01-16 16:41:16 +00:00
return response.json()["success"]