web: Consolidate code to deliver files

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

#!/usr/bin/python

import PIL
import io
import ldap
import logging
import os.path
import tornado.web

from .. import countries
from .. import util

from . import base
from . import ui_modules

COLOUR_LIGHT = (237,232,232)
COLOUR_DARK  = (49,53,60)

class IndexHandler(base.BaseHandler):
	@tornado.web.authenticated
	def get(self):
		results = None

		# Query Term
		q = self.get_argument("q", None)

		# Peform search
		if q:
			results = self.backend.accounts.search(q)

		self.render("users/index.html", q=q, results=results)


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

		# Fetch SIP channels
		sip_channels = await account.get_sip_channels()

		self.render("users/show.html", account=account, sip_channels=sip_channels)


class AvatarHandler(base.BaseHandler):
	async def get(self, uid):
		if self.browser_accepts("image/webp"):
			format = "WEBP"
		else:
			format = "JPEG"

		# Get the desired size of the avatar file
		size = self.get_argument("size", None)

		try:
			size = int(size)
		except (TypeError, ValueError):
			size = None

		logging.debug("Querying for avatar of %s" % uid)

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

		# Allow downstream to cache this for a year
		self.set_expires(31536000)

		# Resize avatar
		avatar = await account.get_avatar(size, format=format)

		# If there is no avatar, we serve a default image
		if not avatar:
			logging.debug("No avatar uploaded for %s" % account)

			# Generate a random avatar with only one letter
			avatar = await self._get_avatar(account, size=size, format=format)

		# Deliver the data
		self._deliver_file(avatar, prefix=account.uid)

	async def _get_avatar(self, account, size=None, format=None, **args):
		letters = account.initials

		if size is None:
			size = 256

		# The generated avatar cannot be larger than 1024px
		if size >= 2048:
			size = 2048

		# Cache key
		key = "avatar:letter:%s:%s:%s" % ("".join(letters), format or "N/A", size)

		# Fetch avatar from the cache
		avatar = await self.backend.cache.get(key)
		if not avatar:
			avatar = self._make_avatar(letters, size=size, **args)

			# Cache for forever
			await self.backend.cache.set(key, avatar)

		return avatar

	def _make_avatar(self, letters, format="PNG", size=None, **args):
		# Load font
		font = PIL.ImageFont.truetype(os.path.join(
			self.application.settings.get("static_path", ""),
			"fonts/Prompt-Bold.ttf"
		), size, encoding="unic")

		width = 0
		height = 0

		for letter in letters:
			# Determine size of the printed letter
			w, h = font.getsize(letter)

			# Store the maximum height
			height = max(h, height)

			# Add up the width
			width += w

		# Add the margin
		width  = int(width  * 1.5)
		height = int(height * 1.5)

		# Generate an image of the correct size
		image = PIL.Image.new("RGBA", (width, height), COLOUR_LIGHT)

		# Have a canvas
		draw = PIL.ImageDraw.Draw(image)

		# Draw the letter in the center
		draw.text((width // 2, height // 2), "".join(letters),
			font=font, anchor="mm", fill=COLOUR_DARK)

		return util.generate_thumbnail(image, size, square=True, format=format)


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

		# Check for permissions
		if not account.can_be_managed_by(self.current_user):
			raise tornado.web.HTTPError(403, "%s cannot manage %s" % (self.current_user, account))

		self.render("users/edit.html", account=account, countries=countries.get_all())

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

		# Check for permissions
		if not account.can_be_managed_by(self.current_user):
			raise tornado.web.HTTPError(403, "%s cannot manage %s" % (self.current_user, account))

		# Unfortunately this cannot be wrapped into a transaction
		try:
			account.first_name   = self.get_argument("first_name")
			account.last_name    = self.get_argument("last_name")
			account.nickname     = self.get_argument("nickname", None)
			account.street       = self.get_argument("street", None)
			account.city         = self.get_argument("city", None)
			account.postal_code  = self.get_argument("postal_code", None)
			account.country_code = self.get_argument("country_code")
			account.description  = self.get_argument("description", None)

			# Avatar
			try:
				filename, data, mimetype = self.get_file("avatar")

				if not mimetype.startswith("image/"):
					raise tornado.web.HTTPError(400, "Avatar is not an image file: %s" % mimetype)

				await account.upload_avatar(data)
			except TypeError:
				pass

			# Email
			account.mail_routing_address = self.get_argument("mail_routing_address", None)

			# Telephone
			account.phone_numbers = self.get_argument("phone_numbers", "").splitlines()
			account.sip_routing_address = self.get_argument("sip_routing_address", None)
		except ldap.STRONG_AUTH_REQUIRED as e:
			raise tornado.web.HTTPError(403, "%s" % e) from e

		# Redirect back to user page
		self.redirect("/users/%s" % account.uid)


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

		# Check for permissions
		if not account.can_be_deleted_by(self.current_user):
			raise tornado.web.HTTPError(403, "%s cannot delete %s" % (self.current_user, account))

		self.render("users/delete.html", account=account)

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

		# Check for permissions
		if not account.can_be_deleted_by(self.current_user):
			raise tornado.web.HTTPError(403, "%s cannot delete %s" % (self.current_user, account))

		# Delete!
		with self.db.transaction():
			await account.delete(self.current_user)

		self.render("users/deleted.html", account=account)


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

		# Check for permissions
		if not account.can_be_managed_by(self.current_user):
			raise tornado.web.HTTPError(403, "%s cannot manage %s" % (self.current_user, account))

		self.render("users/passwd.html", account=account)

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

		# Check for permissions
		if not account.can_be_managed_by(self.current_user):
			raise tornado.web.HTTPError(403, "%s cannot manage %s" % (self.current_user, account))

		# Get current password
		password = self.get_argument("password")

		# Get new password
		password1 = self.get_argument("password1")
		password2 = self.get_argument("password2")

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

		# XXX Check password complexity

		# Check if old password matches
		if not account.check_password(password):
			raise tornado.web.HTTPError(403, "Incorrect password for %s" % account)

		# Save new password
		account.passwd(password1)

		# Redirect back to user's page
		self.redirect("/users/%s" % account.uid)


class GroupIndexHandler(base.BaseHandler):
	@tornado.web.authenticated
	def get(self):
		# Only staff can see other groups
		if not self.current_user.is_staff():
			raise tornado.web.HTTPError(403)

		self.render("users/groups/index.html")


class GroupShowHandler(base.BaseHandler):
	@tornado.web.authenticated
	def get(self, gid):
		# Only staff can see other groups
		if not self.current_user.is_staff():
			raise tornado.web.HTTPError(403)

		# Fetch group
		group = self.backend.groups.get_by_gid(gid)
		if not group:
			raise tornado.web.HTTPError(404, "Could not find group %s" % gid)

		self.render("users/groups/show.html", group=group)


class SubscribeHandler(base.BaseHandler):
	@tornado.web.authenticated
	def get(self):
		self.render("users/subscribe.html")

	@tornado.web.authenticated
	def post(self):
		# Give consent
		with self.db.transaction():
			self.current_user.consents_to_promotional_emails = True

		self.render("users/subscribed.html")


class UnsubscribeHandler(base.BaseHandler):
	@tornado.web.authenticated
	def get(self):
		if self.current_user.consents_to_promotional_emails:
			return self.render("users/unsubscribe.html")

		self.render("users/unsubscribed.html")

	@tornado.web.authenticated
	def post(self):
		# Withdraw consent
		with self.db.transaction():
			self.current_user.consents_to_promotional_emails = False

		self.render("users/unsubscribed.html")


class ListModule(ui_modules.UIModule):
	def render(self, accounts, show_created_at=False):
		return self.render_string("users/modules/list.html", accounts=accounts,
			show_created_at=show_created_at)