[ipfire.org.git] / src / web / auth.py

#!/usr/bin/python

import logging
import tornado.web
import urllib.parse

from . import base
from . import ui_modules

class AuthenticationMixin(object):
	def login(self, account):
		# User has logged in, create a session
		session_id, session_expires = self.backend.accounts.create_session(
			account, self.request.host)

		# Check if a new session was created
		if not session_id:
			raise tornado.web.HTTPError(500, "Could not create session")

		# Send session cookie to the client
		self.set_cookie("session_id", session_id,
			domain=self.request.host, expires=session_expires, secure=True)

	def logout(self):
		session_id = self.get_cookie("session_id")
		if not session_id:
			return

		success = self.backend.accounts.destroy_session(session_id, self.request.host)
		if success:
			self.clear_cookie("session_id")


class LoginHandler(base.AnalyticsMixin, AuthenticationMixin, base.BaseHandler):
	def get(self):
		next = self.get_argument("next", None)

		self.render("auth/login.html", next=next,
			incorrect=False, username=None)

	@base.ratelimit(minutes=15, requests=10)
	def post(self):
		username = self.get_argument("username")
		password = self.get_argument("password")
		next = self.get_argument("next", "/")

		# Find user
		account = self.backend.accounts.auth(username, password)
		if not account:
			logging.error("Unknown user or invalid password: %s" % username)

			# Set status to 401
			self.set_status(401)

			# Render login page again
			return self.render("auth/login.html",
				incorrect=True, username=username, next=next,
			)

		# Create session
		with self.db.transaction():
			self.login(account)

		# Redirect the user
		return self.redirect(next)


class LogoutHandler(AuthenticationMixin, base.BaseHandler):
	def get(self):
		with self.db.transaction():
			self.logout()

		# Get back to the start page
		self.redirect("/")


class JoinHandler(base.AnalyticsMixin, base.BaseHandler):
	def get(self):
		# Redirect logged in users away
		if self.current_user:
			self.redirect("/")
			return

		self.render("auth/join.html")

	@base.ratelimit(minutes=15, requests=5)
	async def post(self):
		uid   = self.get_argument("uid")
		email = self.get_argument("email")

		first_name = self.get_argument("first_name")
		last_name  = self.get_argument("last_name")

		# If the honey field has been set, we probably have a bot
		honey = self.get_argument("honey", None)
		if honey:
			raise tornado.web.HTTPError(503)

		# Fail if first name and last name match
		# Some bots don't seem to be very creative when filling in the form
		if first_name == last_name:
			raise tornado.web.HTTPError(503)

		# Fail if the email address isn't valid
		if self.backend.accounts.mail_is_spam(email):
			raise tornado.web.HTTPError(503, "Email address looks spammy")

		# Register account
		try:
			with self.db.transaction():
				self.backend.accounts.join(uid, email,
					first_name=first_name, last_name=last_name,
					country_code=self.current_country_code)
		except ValueError as e:
			raise tornado.web.HTTPError(400, "%s" % e) from e

		self.render("auth/join-success.html")


class ActivateHandler(AuthenticationMixin, base.BaseHandler):
	def get(self, uid, activation_code):
		self.render("auth/activate.html")

	def post(self, uid, activation_code):
		password1 = self.get_argument("password1")
		password2 = self.get_argument("password2")

		if not password1 == password2:
			raise tornado.web.HTTPError(400, "Passwords do not match")

		with self.db.transaction():
			account = self.backend.accounts.activate(uid, activation_code)
			if not account:
				raise tornado.web.HTTPError(400, "Account not found: %s" % uid)

			# Set the new password
			account.passwd(password1)

			# Create session
			self.login(account)

		# Redirect to success page
		self.render("auth/activated.html", account=account)


class DebugActivatedHandler(base.BaseHandler):
	@tornado.web.authenticated
	def get(self):
		self.render("auth/activated.html", account=self.current_user)


class PasswordResetInitiationHandler(base.BaseHandler):
	def get(self):
		username = self.get_argument("username", None)

		self.render("auth/password-reset-initiation.html", username=username)

	@base.ratelimit(minutes=15, requests=10)
	def post(self):
		username = self.get_argument("username")

		# Fetch account and submit password reset
		with self.db.transaction():
			account = self.backend.accounts.find_account(username)
			if account:
				account.request_password_reset()

		self.render("auth/password-reset-successful.html")


class PasswordResetHandler(AuthenticationMixin, base.BaseHandler):
	def get(self, uid, reset_code):
		account = self.backend.accounts.get_by_uid(uid)
		if not account:
			raise tornado.web.HTTPError(404, "Could not find account: %s" % uid)

		self.render("auth/password-reset.html", account=account)

	def post(self, uid, reset_code):
		account = self.backend.accounts.get_by_uid(uid)
		if not account:
			raise tornado.web.HTTPError(404, "Could not find account: %s" % uid)

		password1 = self.get_argument("password1")
		password2 = self.get_argument("password2")

		if not password1 == password2:
			raise tornado.web.HTTPError(400, "Passwords do not match")

		# Try to perform password reset
		with self.db.transaction():
			account.reset_password(reset_code, password1)

			# Login the user straight away after reset was successful
			self.login(account)

		# Redirect back to /
		self.redirect("/")


class ConfirmEmailHandler(base.BaseHandler):
	def get(self, uid, token):
		account = self.backend.accounts.get_by_uid(uid)
		if not account:
			raise tornado.web.HTTPError(404, "Could not find account: %s" % uid)

		with self.db.transaction():
			success = account.confirm_email(token)

		self.render("auth/email-confirmed.html", success=success)


class WellKnownChangePasswordHandler(base.BaseHandler):
	@tornado.web.authenticated
	def get(self):
		"""
			Implements https://web.dev/articles/change-password-url
		"""
		self.redirect("/users/%s/passwd" % self.current_user.uid)


class SSODiscourse(base.BaseHandler):
	@base.ratelimit(minutes=24*60, requests=100)
	@tornado.web.authenticated
	def get(self):
		# Fetch Discourse's parameters
		sso = self.get_argument("sso")
		sig = self.get_argument("sig")

		# Decode payload
		try:
			params = self.accounts.decode_discourse_payload(sso, sig)

		# Raise bad request if the signature is invalid
		except ValueError:
			raise tornado.web.HTTPError(400)

		# Redirect back if user is already logged in
		args = {
			"nonce" : params.get("nonce"),
			"external_id" : self.current_user.uid,

			# Pass email address
			"email" : self.current_user.email,
			"require_activation" : "false",

			# More details about the user
			"username" : self.current_user.uid,
			"name" : "%s" % self.current_user,
			"bio" : self.current_user.description or "",

			# Avatar
			"avatar_url" : self.current_user.avatar_url(absolute=True),
			"avatar_force_update" : "true",

			# Send a welcome message
			"suppress_welcome_message" : "false",

			# Group memberships
			"groups" : ",".join((group.gid for group in self.current_user.groups)),

			# Admin?
			"admin" : "true" if self.current_user.is_admin() else "false",

			# Moderator?
			"moderator" : "true" if self.current_user.is_moderator() else "false",
		}

		# Format payload and sign it
		payload = self.accounts.encode_discourse_payload(**args)
		signature = self.accounts.sign_discourse_payload(payload)

		qs = urllib.parse.urlencode({
			"sso" : payload,
			"sig" : signature,
		})

		# Redirect user
		self.redirect("%s?%s" % (params.get("return_sso_url"), qs))


class PasswordModule(ui_modules.UIModule):
	def render(self, account=None):
		return self.render_string("auth/modules/password.html", account=account)

	def javascript_files(self):
		return "js/zxcvbn.js"

	def embedded_javascript(self):
		return self.render_string("auth/modules/password.js")


class APICheckUID(base.APIHandler):
	@base.ratelimit(minutes=1, requests=100)
	def get(self):
		uid = self.get_argument("uid")
		result = None

		if not uid:
			result = "empty"

		# Check if the username is syntactically valid
		elif not self.backend.accounts.uid_is_valid(uid):
			result = "invalid"

		# Check if the username is already taken
		elif self.backend.accounts.uid_exists(uid):
			result = "taken"

		# Username seems to be okay
		self.finish({ "result" : result or "ok" })


class APICheckEmail(base.APIHandler):
	@base.ratelimit(minutes=1, requests=100)
	def get(self):
		email = self.get_argument("email")
		result = None

		if not email:
			result = "empty"

		elif not self.backend.accounts.mail_is_valid(email):
			result = "invalid"

		# Check if this email address is blacklisted
		elif self.backend.accounts.mail_is_blacklisted(email):
			result = "blacklisted"

		# Check if this email address is already useed
		elif self.backend.accounts.get_by_mail(email):
			result = "taken"

		self.finish({ "result" : result or "ok" })
Commit	Line	Data
	1	#!/usr/bin/python
	2
	3	import logging
	4	import tornado.web
	5	import urllib.parse
	6
	7	from . import base
	8	from . import ui_modules
	9
	10	class AuthenticationMixin(object):
	11	def login(self, account):
	12	# User has logged in, create a session
	13	session_id, session_expires = self.backend.accounts.create_session(
	14	account, self.request.host)
	15
	16	# Check if a new session was created
	17	if not session_id:
	18	raise tornado.web.HTTPError(500, "Could not create session")
	19
	20	# Send session cookie to the client
	21	self.set_cookie("session_id", session_id,
	22	domain=self.request.host, expires=session_expires, secure=True)
	23
	24	def logout(self):
	25	session_id = self.get_cookie("session_id")
	26	if not session_id:
	27	return
	28
	29	success = self.backend.accounts.destroy_session(session_id, self.request.host)
	30	if success:
	31	self.clear_cookie("session_id")
	32
	33
	34	class LoginHandler(base.AnalyticsMixin, AuthenticationMixin, base.BaseHandler):
	35	def get(self):
	36	next = self.get_argument("next", None)
	37
	38	self.render("auth/login.html", next=next,
	39	incorrect=False, username=None)
	40
	41	@base.ratelimit(minutes=15, requests=10)
	42	def post(self):
	43	username = self.get_argument("username")
	44	password = self.get_argument("password")
	45	next = self.get_argument("next", "/")
	46
	47	# Find user
	48	account = self.backend.accounts.auth(username, password)
	49	if not account:
	50	logging.error("Unknown user or invalid password: %s" % username)
	51
	52	# Set status to 401
	53	self.set_status(401)
	54
	55	# Render login page again
	56	return self.render("auth/login.html",
	57	incorrect=True, username=username, next=next,
	58	)
	59
	60	# Create session
	61	with self.db.transaction():
	62	self.login(account)
	63
	64	# Redirect the user
	65	return self.redirect(next)
	66
	67
	68	class LogoutHandler(AuthenticationMixin, base.BaseHandler):
	69	def get(self):
	70	with self.db.transaction():
	71	self.logout()
	72
	73	# Get back to the start page
	74	self.redirect("/")
	75
	76
	77	class JoinHandler(base.AnalyticsMixin, base.BaseHandler):
	78	def get(self):
	79	# Redirect logged in users away
	80	if self.current_user:
	81	self.redirect("/")
	82	return
	83
	84	self.render("auth/join.html")
	85
	86	@base.ratelimit(minutes=15, requests=5)
	87	async def post(self):
	88	uid = self.get_argument("uid")
	89	email = self.get_argument("email")
	90
	91	first_name = self.get_argument("first_name")
	92	last_name = self.get_argument("last_name")
	93
	94	# If the honey field has been set, we probably have a bot
	95	honey = self.get_argument("honey", None)
	96	if honey:
	97	raise tornado.web.HTTPError(503)
	98
	99	# Fail if first name and last name match
	100	# Some bots don't seem to be very creative when filling in the form
	101	if first_name == last_name:
	102	raise tornado.web.HTTPError(503)
	103
	104	# Fail if the email address isn't valid
	105	if self.backend.accounts.mail_is_spam(email):
	106	raise tornado.web.HTTPError(503, "Email address looks spammy")
	107
	108	# Register account
	109	try:
	110	with self.db.transaction():
	111	self.backend.accounts.join(uid, email,
	112	first_name=first_name, last_name=last_name,
	113	country_code=self.current_country_code)
	114	except ValueError as e:
	115	raise tornado.web.HTTPError(400, "%s" % e) from e
	116
	117	self.render("auth/join-success.html")
	118
	119
	120	class ActivateHandler(AuthenticationMixin, base.BaseHandler):
	121	def get(self, uid, activation_code):
	122	self.render("auth/activate.html")
	123
	124	def post(self, uid, activation_code):
	125	password1 = self.get_argument("password1")
	126	password2 = self.get_argument("password2")
	127
	128	if not password1 == password2:
	129	raise tornado.web.HTTPError(400, "Passwords do not match")
	130
	131	with self.db.transaction():
	132	account = self.backend.accounts.activate(uid, activation_code)
	133	if not account:
	134	raise tornado.web.HTTPError(400, "Account not found: %s" % uid)
	135
	136	# Set the new password
	137	account.passwd(password1)
	138
	139	# Create session
	140	self.login(account)
	141
	142	# Redirect to success page
	143	self.render("auth/activated.html", account=account)
	144
	145
	146	class DebugActivatedHandler(base.BaseHandler):
	147	@tornado.web.authenticated
	148	def get(self):
	149	self.render("auth/activated.html", account=self.current_user)
	150
	151
	152	class PasswordResetInitiationHandler(base.BaseHandler):
	153	def get(self):
	154	username = self.get_argument("username", None)
	155
	156	self.render("auth/password-reset-initiation.html", username=username)
	157
	158	@base.ratelimit(minutes=15, requests=10)
	159	def post(self):
	160	username = self.get_argument("username")
	161
	162	# Fetch account and submit password reset
	163	with self.db.transaction():
	164	account = self.backend.accounts.find_account(username)
	165	if account:
	166	account.request_password_reset()
	167
	168	self.render("auth/password-reset-successful.html")
	169
	170
	171	class PasswordResetHandler(AuthenticationMixin, base.BaseHandler):
	172	def get(self, uid, reset_code):
	173	account = self.backend.accounts.get_by_uid(uid)
	174	if not account:
	175	raise tornado.web.HTTPError(404, "Could not find account: %s" % uid)
	176
	177	self.render("auth/password-reset.html", account=account)
	178
	179	def post(self, uid, reset_code):
	180	account = self.backend.accounts.get_by_uid(uid)
	181	if not account:
	182	raise tornado.web.HTTPError(404, "Could not find account: %s" % uid)
	183
	184	password1 = self.get_argument("password1")
	185	password2 = self.get_argument("password2")
	186
	187	if not password1 == password2:
	188	raise tornado.web.HTTPError(400, "Passwords do not match")
	189
	190	# Try to perform password reset
	191	with self.db.transaction():
	192	account.reset_password(reset_code, password1)
	193
	194	# Login the user straight away after reset was successful
	195	self.login(account)
	196
	197	# Redirect back to /
	198	self.redirect("/")
	199
	200
	201	class ConfirmEmailHandler(base.BaseHandler):
	202	def get(self, uid, token):
	203	account = self.backend.accounts.get_by_uid(uid)
	204	if not account:
	205	raise tornado.web.HTTPError(404, "Could not find account: %s" % uid)
	206
	207	with self.db.transaction():
	208	success = account.confirm_email(token)
	209
	210	self.render("auth/email-confirmed.html", success=success)
	211
	212
	213	class WellKnownChangePasswordHandler(base.BaseHandler):
	214	@tornado.web.authenticated
	215	def get(self):
	216	"""
	217	Implements https://web.dev/articles/change-password-url
	218	"""
	219	self.redirect("/users/%s/passwd" % self.current_user.uid)
	220
	221
	222	class SSODiscourse(base.BaseHandler):
	223	@base.ratelimit(minutes=24*60, requests=100)
	224	@tornado.web.authenticated
	225	def get(self):
	226	# Fetch Discourse's parameters
	227	sso = self.get_argument("sso")
	228	sig = self.get_argument("sig")
	229
	230	# Decode payload
	231	try:
	232	params = self.accounts.decode_discourse_payload(sso, sig)
	233
	234	# Raise bad request if the signature is invalid
	235	except ValueError:
	236	raise tornado.web.HTTPError(400)
	237
	238	# Redirect back if user is already logged in
	239	args = {
	240	"nonce" : params.get("nonce"),
	241	"external_id" : self.current_user.uid,
	242
	243	# Pass email address
	244	"email" : self.current_user.email,
	245	"require_activation" : "false",
	246
	247	# More details about the user
	248	"username" : self.current_user.uid,
	249	"name" : "%s" % self.current_user,
	250	"bio" : self.current_user.description or "",
	251
	252	# Avatar
	253	"avatar_url" : self.current_user.avatar_url(absolute=True),
	254	"avatar_force_update" : "true",
	255
	256	# Send a welcome message
	257	"suppress_welcome_message" : "false",
	258
	259	# Group memberships
	260	"groups" : ",".join((group.gid for group in self.current_user.groups)),
	261
	262	# Admin?
	263	"admin" : "true" if self.current_user.is_admin() else "false",
	264
	265	# Moderator?
	266	"moderator" : "true" if self.current_user.is_moderator() else "false",
	267	}
	268
	269	# Format payload and sign it
	270	payload = self.accounts.encode_discourse_payload(**args)
	271	signature = self.accounts.sign_discourse_payload(payload)
	272
	273	qs = urllib.parse.urlencode({
	274	"sso" : payload,
	275	"sig" : signature,
	276	})
	277
	278	# Redirect user
	279	self.redirect("%s?%s" % (params.get("return_sso_url"), qs))
	280
	281
	282	class PasswordModule(ui_modules.UIModule):
	283	def render(self, account=None):
	284	return self.render_string("auth/modules/password.html", account=account)
	285
	286	def javascript_files(self):
	287	return "js/zxcvbn.js"
	288
	289	def embedded_javascript(self):
	290	return self.render_string("auth/modules/password.js")
	291
	292
	293	class APICheckUID(base.APIHandler):
	294	@base.ratelimit(minutes=1, requests=100)
	295	def get(self):
	296	uid = self.get_argument("uid")
	297	result = None
	298
	299	if not uid:
	300	result = "empty"
	301
	302	# Check if the username is syntactically valid
	303	elif not self.backend.accounts.uid_is_valid(uid):
	304	result = "invalid"
	305
	306	# Check if the username is already taken
	307	elif self.backend.accounts.uid_exists(uid):
	308	result = "taken"
	309
	310	# Username seems to be okay
	311	self.finish({ "result" : result or "ok" })
	312
	313
	314	class APICheckEmail(base.APIHandler):
	315	@base.ratelimit(minutes=1, requests=100)
	316	def get(self):
	317	email = self.get_argument("email")
	318	result = None
	319
	320	if not email:
	321	result = "empty"
	322
	323	elif not self.backend.accounts.mail_is_valid(email):
	324	result = "invalid"
	325
	326	# Check if this email address is blacklisted
	327	elif self.backend.accounts.mail_is_blacklisted(email):
	328	result = "blacklisted"
	329
	330	# Check if this email address is already useed
	331	elif self.backend.accounts.get_by_mail(email):
	332	result = "taken"
	333
	334	self.finish({ "result" : result or "ok" })