accounts: Drop StopForumSpam

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

#!/usr/bin/python

import logging
import tornado.web
import urllib.parse

from . import base

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(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 RegisterHandler(base.BaseHandler):
	def get(self):
		# Redirect logged in users away
		if self.current_user:
			self.redirect("/")
			return

		self.render("auth/register.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")

		# Register account
		try:
			with self.db.transaction():
				self.backend.accounts.register(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/register-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 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
		account = self.backend.accounts.get_by_uid(username)
		if account:
			with self.db.transaction():
				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 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
			"admin" : "true" if self.current_user.is_admin() else "false",
			"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 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" })