users: Add index page

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

#!/usr/bin/python

import PIL
import datetime
import imghdr
import io
import ldap
import logging
import os.path
import tornado.web
import urllib.parse

from .. import countries

from . import auth
from . import base
from . import ui_modules

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

class IndexHandler(auth.CacheMixin, base.BaseHandler):
	@tornado.web.authenticated
	def get(self):
		self.render("people/index.html")


class AvatarHandler(base.BaseHandler):
	def get(self, uid):
		# 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 = account.get_avatar(size)

		# 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 = self._get_avatar(account, size=size)

		# Guess content type
		type = imghdr.what(None, avatar)

		# Set headers about content
		self.set_header("Content-Disposition", "inline; filename=\"%s.%s\"" % (account.uid, type))
		self.set_header("Content-Type", "image/%s" % type)

		# Deliver payload
		self.finish(avatar)

	def _get_avatar(self, account, size=None, **args):
		letter = ("%s" % account)[0].upper()

		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" % (letter, size)

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

			# Cache for forever
			self.memcached.set(key, avatar)

		return avatar

	def _make_avatar(self, letter, format="PNG", size=None, **args):
		# Generate an image of the correct size
		image = PIL.Image.new("RGBA", (size, size), COLOUR_LIGHT)

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

		# Load font
		font = PIL.ImageFont.truetype(os.path.join(
			self.application.settings.get("static_path", ""),
			"fonts/Prompt-Bold.ttf"
		), size, encoding="unic")

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

		# Mukta seems to be very broken and the height needs to be corrected
		h //= 0.7

		# Draw the letter in the center
		draw.text(((size - w) / 2, (size - h) / 2), letter,
			font=font, fill=COLOUR_DARK)

		with io.BytesIO() as f:
			# If writing out the image does not work with optimization,
			# we try to write it out without any optimization.
			try:
				image.save(f, format, optimize=True, **args)
			except:
				image.save(f, format, **args)

			return f.getvalue()


class CallsHandler(auth.CacheMixin, base.BaseHandler):
	@tornado.web.authenticated
	def get(self, uid, date=None):
		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))

		if date:
			try:
				date = datetime.datetime.strptime(date, "%Y-%m-%d").date()
			except ValueError:
				raise tornado.web.HTTPError(400, "Invalid date: %s" % date)
		else:
			date = datetime.date.today()

		self.render("people/calls.html", account=account, date=date)


class CallHandler(auth.CacheMixin, base.BaseHandler):
	@tornado.web.authenticated
	def get(self, uid, uuid):
		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))

		call = self.backend.talk.freeswitch.get_call_by_uuid(uuid)
		if not call:
			raise tornado.web.HTTPError(404, "Could not find call %s" % uuid)

		# XXX limit

		self.render("people/call.html", account=account, call=call)


class ConferencesHandler(auth.CacheMixin, base.BaseHandler):
	@tornado.web.authenticated
	def get(self):
		self.render("people/conferences.html", conferences=self.backend.talk.conferences)


class GroupsHandler(auth.CacheMixin, 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("people/groups.html")


class GroupHandler(auth.CacheMixin, 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("people/group.html", group=group)


class SearchHandler(auth.CacheMixin, base.BaseHandler):
	@tornado.web.authenticated
	def get(self):
		q = self.get_argument("q")

		# Perform the search
		accounts = self.backend.accounts.search(q)

		# Redirect when only one result was found
		if len(accounts) == 1:
			self.redirect("/users/%s" % accounts[0].uid)
			return

		self.render("people/search.html", q=q, accounts=accounts)


class StatsHandler(auth.CacheMixin, base.BaseHandler):
	@tornado.web.authenticated
	def get(self):
		# Only staff can see stats
		if not self.current_user.is_staff():
			raise tornado.web.HTTPError(403)

		self.render("people/stats.html")


class SubscribeHandler(auth.CacheMixin, base.BaseHandler):
	@tornado.web.authenticated
	def post(self):
		# Give consent
		self.current_user.consents_to_promotional_emails = True

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


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

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

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

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


class SIPHandler(auth.CacheMixin, 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("people/sip.html", account=account)


class UserEditHandler(auth.CacheMixin, 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("people/user-edit.html", account=account, countries=countries.get_all())

	@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))

		# 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", None)
			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)

				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 UserPasswdHandler(auth.CacheMixin, 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("people/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 SSODiscourse(auth.CacheMixin, 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(),
			"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 AccountsListModule(ui_modules.UIModule):
	def render(self, accounts=None):
		if accounts is None:
			accounts = self.backend.accounts

		return self.render_string("people/modules/accounts-list.html", accounts=accounts)


class AgentModule(ui_modules.UIModule):
	def render(self, account):
		return self.render_string("people/modules/agent.html", account=account)


class CDRModule(ui_modules.UIModule):
	def render(self, account, date=None, limit=None):
		cdr = account.get_cdr(date=date, limit=limit)

		return self.render_string("people/modules/cdr.html",
			account=account, cdr=list(cdr))


class ChannelsModule(ui_modules.UIModule):
	def render(self, account):
		return self.render_string("people/modules/channels.html",
			account=account, channels=account.sip_channels)


class MOSModule(ui_modules.UIModule):
	def render(self, call):
		return self.render_string("people/modules/mos.html", call=call)


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

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

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


class RegistrationsModule(ui_modules.UIModule):
	def render(self, account):
		return self.render_string("people/modules/registrations.html", account=account)