import os
import hashlib
import datetime
import time
import uuid
import hmac
import base64
import requests
import simplejson as json
from flask import (Flask, g, render_template, url_for, request, make_response,
redirect, flash, abort)
from gn2.wqflask import app
from gn2.wqflask import pbkdf2
from gn2.wqflask.user_session import UserSession
from gn2.utility import hmac
from gn2.utility.redis_tools import is_redis_available, get_redis_conn, get_user_id, get_user_by_unique_column, set_user_attribute, save_user, save_verification_code, check_verification_code, get_user_collections, save_collections
Redis = get_redis_conn()
from smtplib import SMTP
from gn2.utility.tools import SMTP_CONNECT, SMTP_USERNAME, SMTP_PASSWORD, LOG_SQL_ALCHEMY, GN2_BRANCH_URL
THREE_DAYS = 60 * 60 * 24 * 3
def timestamp():
return datetime.datetime.utcnow().isoformat()
def basic_info():
return dict(timestamp=timestamp(),
ip_address=request.remote_addr,
user_agent=request.headers.get('User-Agent'))
def encode_password(pass_gen_fields, unencrypted_password):
if isinstance(pass_gen_fields['salt'], bytes):
salt = pass_gen_fields['salt']
else:
salt = bytes(pass_gen_fields['salt'], "utf-8")
encrypted_password = pbkdf2.pbkdf2_hex(str(unencrypted_password),
salt,
pass_gen_fields['iterations'],
pass_gen_fields['keylength'],
pass_gen_fields['hashfunc'])
pass_gen_fields.pop("unencrypted_password", None)
pass_gen_fields["password"] = encrypted_password
return pass_gen_fields
def set_password(password):
pass_gen_fields = {
"unencrypted_password": password,
"algorithm": "pbkdf2",
"hashfunc": "sha256",
"salt": base64.b64encode(os.urandom(32)),
"iterations": 100000,
"keylength": 32,
"created_timestamp": timestamp()
}
assert len(password) >= 6, "Password shouldn't be shorter than 6 characters"
encoded_password = encode_password(
pass_gen_fields, pass_gen_fields['unencrypted_password'])
return encoded_password
def get_signed_session_id(user):
session_id = str(uuid.uuid4())
session_id_signature = hmac.hmac_creation(session_id)
session_id_signed = session_id + ":" + session_id_signature
# ZS: Need to check if this is ever actually used or exists
if 'user_id' not in user:
user['user_id'] = str(uuid.uuid4())
save_user(user, user['user_id'])
if 'github_id' in user:
session = dict(login_time=time.time(),
user_type="github",
user_id=user['user_id'],
github_id=user['github_id'],
user_name=user['name'],
user_url=user['user_url'])
elif 'orcid' in user:
session = dict(login_time=time.time(),
user_type="orcid",
user_id=user['user_id'],
github_id=user['orcid'],
user_name=user['name'],
user_url=user['user_url'])
else:
session = dict(login_time=time.time(),
user_type="gn2",
user_id=user['user_id'],
user_name=user['full_name'],
user_email_address=user['email_address'])
key = UserSession.user_cookie_name + ":" + session_id
Redis.hmset(key, session)
Redis.expire(key, THREE_DAYS)
return session_id_signed
def send_email(toaddr, msg, fromaddr="no-reply@genenetwork.org"):
"""Send an E-mail through SMTP_CONNECT host. If SMTP_USERNAME is not
'UNKNOWN' TLS is used
"""
if SMTP_USERNAME == 'UNKNOWN':
server = SMTP(SMTP_CONNECT)
server.sendmail(fromaddr, toaddr, msg)
else:
server = SMTP(SMTP_CONNECT)
server.starttls()
server.login(SMTP_USERNAME, SMTP_PASSWORD)
server.sendmail(fromaddr, toaddr, msg)
server.quit()
def send_verification_email(user_details, template_name="email/user_verification.txt", key_prefix="verification_code", subject="GeneNetwork e-mail verification"):
verification_code = str(uuid.uuid4())
key = key_prefix + ":" + verification_code
data = json.dumps(dict(id=user_details['user_id'], timestamp=timestamp()))
Redis.set(key, data)
Redis.expire(key, THREE_DAYS)
recipient = user_details['email_address']
body = render_template(template_name, verification_code=verification_code)
send_email(recipient, subject, body)
return {"recipient": recipient, "subject": subject, "body": body}
def send_invitation_email(user_email, temp_password, template_name="email/user_invitation.txt", subject="You've been added to a GeneNetwork user group"):
recipient = user_email
body = render_template(template_name, temp_password)
send_email(recipient, subject, body)
return {"recipient": recipient, "subject": subject, "body": body}
@app.route("/manage/verify_email")
def verify_email():
if 'code' in request.args:
user_details = check_verification_code(request.args['code'])
if user_details:
# As long as they have access to the email account
# We might as well log them in
session_id_signed = get_signed_session_id(user_details)
flash("Thank you for logging in {}.".format(
user_details['full_name']), "alert-success")
response = make_response(redirect(
url_for('index_page', import_collections=import_col, anon_id=anon_id)))
response.set_cookie(UserSession.user_cookie_name,
session_id_signed, max_age=None)
return response
else:
flash(
"Invalid code: Password reset code does not exist or might have expired!", "error")
@app.route("/n/login", methods=('GET', 'POST'))
def login():
params = request.form if request.form else request.args
if not params: # ZS: If coming to page for first time
from gn2.utility.tools import GITHUB_AUTH_URL, GITHUB_CLIENT_ID, ORCID_AUTH_URL, ORCID_CLIENT_ID
external_login = {}
if GITHUB_AUTH_URL and GITHUB_CLIENT_ID != 'UNKNOWN':
external_login["github"] = GITHUB_AUTH_URL
if ORCID_AUTH_URL and ORCID_CLIENT_ID != 'UNKNOWN':
external_login["orcid"] = ORCID_AUTH_URL
return render_template("new_security/login_user.html", external_login=external_login, redis_is_available=is_redis_available())
else: # ZS: After clicking sign-in
if 'type' in params and 'uid' in params:
user_details = get_user_by_unique_column("user_id", params['uid'])
if user_details:
session_id_signed = get_signed_session_id(user_details)
if 'name' in user_details and user_details['name'] != "None":
display_id = user_details['name']
elif 'github_id' in user_details:
display_id = user_details['github_id']
elif 'orcid' in user_details:
display_id = user_details['orcid']
else:
display_id = ""
flash("Thank you for logging in {}.".format(
display_id), "alert-success")
response = make_response(redirect(url_for('index_page')))
response.set_cookie(
UserSession.user_cookie_name, session_id_signed, max_age=None)
else:
flash("Something went unexpectedly wrong.", "alert-danger")
response = make_response(redirect(url_for('index_page')))
return response
else:
user_details = get_user_by_unique_column(
"email_address", params['email_address'])
password_match = False
if user_details:
submitted_password = params['password']
pwfields = user_details['password']
if isinstance(pwfields, str):
pwfields = json.loads(pwfields)
encrypted_pass_fields = encode_password(
pwfields, submitted_password)
password_match = pbkdf2.safe_str_cmp(
encrypted_pass_fields['password'], pwfields['password'])
else: # Invalid e-mail
flash("Invalid e-mail address. Please try again.", "alert-danger")
response = make_response(redirect(url_for('login')))
return response
if password_match: # If password correct
if user_details['confirmed']: # If account confirmed
import_col = "false"
anon_id = ""
if 'import_collections' in params:
import_col = "true"
anon_id = params['anon_id']
session_id_signed = get_signed_session_id(user_details)
flash("Thank you for logging in {}.".format(
user_details['full_name']), "alert-success")
response = make_response(redirect(
url_for('index_page', import_collections=import_col, anon_id=anon_id)))
response.set_cookie(
UserSession.user_cookie_name, session_id_signed, max_age=None)
return response
else:
email_ob = send_verification_email(
user_details, template_name="email/user_verification.txt")
return render_template("newsecurity/verification_still_needed.html", subject=email_ob['subject'])
else: # Incorrect password
# ZS: It previously seemed to store that there was an incorrect log-in attempt here, but it did so in the MySQL DB so this might need to be reproduced with Redis
flash("Invalid password. Please try again.", "alert-danger")
response = make_response(redirect(url_for('login')))
return response
@app.route("/n/login/github_oauth2", methods=('GET', 'POST'))
def github_oauth2():
from gn2.utility.tools import GITHUB_CLIENT_ID, GITHUB_CLIENT_SECRET, GITHUB_AUTH_URL
code = request.args.get("code")
data = {
"client_id": GITHUB_CLIENT_ID,
"client_secret": GITHUB_CLIENT_SECRET,
"code": code
}
result = requests.post(
"https://github.com/login/oauth/access_token", json=data)
result_dict = {arr[0]: arr[1]
for arr in [tok.split("=") for tok in result.text.split("&")]}
github_user = get_github_user_details(result_dict["access_token"])
user_details = get_user_by_unique_column("github_id", github_user["id"])
if user_details == None:
user_details = {
"user_id": str(uuid.uuid4()),
"name": github_user["name"].encode("utf-8") if github_user["name"] else "None",
"github_id": github_user["id"],
"user_url": github_user["html_url"].encode("utf-8"),
"login_type": "github",
"organization": "",
"active": 1,
"confirmed": 1
}
save_user(user_details, user_details["user_id"])
url = "/n/login?type=github&uid=" + user_details["user_id"]
return redirect(url)
def get_github_user_details(access_token):
from gn2.utility.tools import GITHUB_API_URL
result = requests.get(GITHUB_API_URL, headers={
'Authorization': 'token ' + access_token}).content
return json.loads(result)
@app.route("/n/login/orcid_oauth2", methods=('GET', 'POST'))
def orcid_oauth2():
from uuid import uuid4
from gn2.utility.tools import ORCID_CLIENT_ID, ORCID_CLIENT_SECRET, ORCID_TOKEN_URL, ORCID_AUTH_URL
code = request.args.get("code")
error = request.args.get("error")
url = "/n/login"
if code:
data = {
"client_id": ORCID_CLIENT_ID,
"client_secret": ORCID_CLIENT_SECRET,
"grant_type": "authorization_code",
"redirect_uri": GN2_BRANCH_URL + "n/login/orcid_oauth2",
"code": code
}
result = requests.post(ORCID_TOKEN_URL, data=data)
result_dict = json.loads(result.text.encode("utf-8"))
user_details = get_user_by_unique_column("orcid", result_dict["orcid"])
if user_details == None:
user_details = {
"user_id": str(uuid4()),
"name": result_dict["name"],
"orcid": result_dict["orcid"],
"user_url": "%s/%s" % ("/".join(ORCID_AUTH_URL.split("/")[:-2]), result_dict["orcid"]),
"login_type": "orcid",
"organization": "",
"active": 1,
"confirmed": 1
}
save_user(user_details, user_details["user_id"])
url = "/n/login?type=orcid&uid=" + user_details["user_id"]
else:
flash("There was an error getting code from ORCID")
return redirect(url)
def get_github_user_details(access_token):
from gn2.utility.tools import GITHUB_API_URL
result = requests.get(GITHUB_API_URL, headers={
'Authorization': 'token ' + access_token}).content
return json.loads(result)
@app.route("/n/logout")
def logout():
UserSession().delete_session()
flash("You are now logged out. We hope you come back soon!")
response = make_response(redirect(url_for('index_page')))
# Delete the cookie
response.set_cookie(UserSession.user_cookie_name, '', expires=0)
return response
@app.route("/n/forgot_password", methods=['GET'])
def forgot_password():
"""Entry point for forgotten password"""
print("ARGS: ", request.args)
errors = {"no-email": request.args.get("no-email")}
print("ERRORS: ", errors)
return render_template("new_security/forgot_password.html", errors=errors)
def send_forgot_password_email(verification_email):
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
template_name = "email/forgot_password.txt"
key_prefix = "forgot_password_code"
subject = "GeneNetwork password reset"
fromaddr = "no-reply@genenetwork.org"
verification_code = str(uuid.uuid4())
key = key_prefix + ":" + verification_code
data = {
"verification_code": verification_code,
"email_address": verification_email,
"timestamp": timestamp()
}
save_verification_code(verification_email, verification_code)
body = render_template(template_name, verification_code=verification_code)
msg = MIMEMultipart()
msg["To"] = verification_email
msg["Subject"] = subject
msg["From"] = fromaddr
msg.attach(MIMEText(body, "plain"))
send_email(verification_email, msg.as_string())
return subject
@app.route("/n/forgot_password_submit", methods=('POST',))
def forgot_password_submit():
"""When a forgotten password form is submitted we get here"""
params = request.form
email_address = params['email_address']
next_page = None
if email_address != "":
user_details = get_user_by_unique_column(
"email_address", email_address)
if user_details:
email_subject = send_forgot_password_email(
user_details["email_address"])
return render_template("new_security/forgot_password_step2.html",
subject=email_subject)
else:
flash("The e-mail entered is not associated with an account.",
"alert-danger")
return redirect(url_for("forgot_password"))
else:
flash("You MUST provide an email", "alert-danger")
return redirect(url_for("forgot_password"))
@app.route("/n/password_reset", methods=['GET'])
def password_reset():
"""Entry point after user clicks link in E-mail"""
verification_code = request.args.get('code')
hmac = request.args.get('hm')
if verification_code:
user_details = check_verification_code(verification_code)
if user_details:
return render_template(
"new_security/password_reset.html", user_encode=user_details["email_address"])
else:
flash(
"Invalid code: Password reset code does not exist or might have expired!", "error")
return redirect(url_for("login"))
else:
return redirect(url_for("login"))
@app.route("/n/password_reset_step2", methods=('POST',))
def password_reset_step2():
"""Handle confirmation E-mail for password reset"""
errors = []
user_email = request.form['user_encode']
user_id = get_user_id("email_address", user_email)
password = request.form['password']
encoded_password = set_password(password)
set_user_attribute(user_id, "password", encoded_password)
flash("Password changed successfully. You can now sign in.", "alert-info")
return redirect(url_for('login'))
def register_user(params):
thank_you_mode = False
errors = []
user_details = {}
user_details['email_address'] = params.get(
'email_address', '').encode("utf-8").strip()
if not (5 <= len(user_details['email_address']) <= 50):
errors.append(
'Email Address needs to be between 5 and 50 characters.')
else:
email_exists = get_user_by_unique_column(
"email_address", user_details['email_address'])
if email_exists:
errors.append('User already exists with that email')
user_details['full_name'] = params.get(
'full_name', '').encode("utf-8").strip()
if not (5 <= len(user_details['full_name']) <= 50):
errors.append('Full Name needs to be between 5 and 50 characters.')
user_details['organization'] = params.get(
'organization', '').encode("utf-8").strip()
if user_details['organization'] and not (5 <= len(user_details['organization']) <= 50):
errors.append(
'Organization needs to be empty or between 5 and 50 characters.')
password = str(params.get('password', ''))
if not (6 <= len(password)):
errors.append('Password needs to be at least 6 characters.')
if params.get('password_confirm') != password:
errors.append("Passwords don't match.")
user_details['password'] = set_password(password)
user_details['user_id'] = str(uuid.uuid4())
user_details['confirmed'] = 1
user_details['registration_info'] = basic_info()
if len(errors) == 0:
save_user(user_details, user_details['user_id'])
return errors
@app.route("/n/register", methods=('GET', 'POST'))
def register():
errors = []
params = request.form if request.form else request.args
params = params.to_dict(flat=True)
if params:
errors = register_user(params)
if len(errors) == 0:
flash(
"Registration successful. You may login with your new account", "alert-info")
return redirect(url_for("login"))
return render_template("new_security/register_user.html", values=params, errors=errors)
@app.errorhandler(401)
def unauthorized(error):
return redirect(url_for('login'))
