talk: Properly size all icons

[ipfire.org.git] / src / backend / accounts.py

#!/usr/bin/python
# encoding: utf-8

import PIL
import io
import ldap
import logging
import urllib.parse
import urllib.request

from .decorators import *
from .misc import Object

class Accounts(Object):
	def __iter__(self):
		# Only return developers (group with ID 1000)
		accounts = self._search("(&(objectClass=posixAccount)(gidNumber=1000))")

		return iter(sorted(accounts))

	@lazy_property
	def ldap(self):
		# Connect to LDAP server
		ldap_uri = self.settings.get("ldap_uri")
		conn = ldap.initialize(ldap_uri)

		# Bind with username and password
		bind_dn = self.settings.get("ldap_bind_dn")
		if bind_dn:
			bind_pw = self.settings.get("ldap_bind_pw", "")
			conn.simple_bind(bind_dn, bind_pw)

		return conn

	def _query(self, query, attrlist=None, limit=0):
		logging.debug("Performing LDAP query: %s" % query)

		search_base = self.settings.get("ldap_search_base")

		try:
			results = self.ldap.search_ext_s(search_base, ldap.SCOPE_SUBTREE,
				query, attrlist=attrlist, sizelimit=limit)
		except:
			# Close current connection
			self.ldap.close()
			del self.ldap

			raise

		return results

	def _search(self, query, attrlist=None, limit=0):
		accounts = []

		for dn, attrs in self._query(query, attrlist=attrlist, limit=limit):
			account = Account(self.backend, dn, attrs)
			accounts.append(account)

		return accounts

	def search(self, query):
		# Search for exact matches
		accounts = self._search("(&(objectClass=posixAccount) \
			(|(uid=%s)(mail=%s)(sipAuthenticationUser=%s)(telephoneNumber=%s)(homePhone=%s)(mobile=%s)))" \
			% (query, query, query, query, query, query))

		# Find accounts by name
		if not accounts:
			for account in self._search("(&(objectClass=posixAccount)(cn=*%s*))" % query):
				if not account in accounts:
					accounts.append(account)

		return sorted(accounts)

	def _search_one(self, query):
		result = self._search(query, limit=1)
		assert len(result) <= 1

		if result:
			return result[0]

	def get_by_uid(self, uid):
		return self._search_one("(&(objectClass=posixAccount)(uid=%s))" % uid)

	def get_by_mail(self, mail):
		return self._search_one("(&(objectClass=posixAccount)(mail=%s))" % mail)

	find = get_by_uid

	def find_account(self, s):
		account = self.get_by_uid(s)
		if account:
			return account

		return self.get_by_mail(s)

	def get_by_sip_id(self, sip_id):
		return self._search_one("(|(&(objectClass=sipUser)(sipAuthenticationUser=%s)) \
			(&(objectClass=sipRoutingObject)(sipLocalAddress=%s)))" % (sip_id, sip_id))

	# Session stuff

	def _cleanup_expired_sessions(self):
		self.db.execute("DELETE FROM sessions WHERE time_expires <= NOW()")

	def create_session(self, account, host):
		self._cleanup_expired_sessions()

		res = self.db.get("INSERT INTO sessions(host, uid) VALUES(%s, %s) \
			RETURNING session_id, time_expires", host, account.uid)

		# Session could not be created
		if not res:
			return None, None

		logging.info("Created session %s for %s which expires %s" \
			% (res.session_id, account, res.time_expires))
		return res.session_id, res.time_expires

	def destroy_session(self, session_id, host):
		logging.info("Destroying session %s" % session_id)

		self.db.execute("DELETE FROM sessions \
			WHERE session_id = %s AND host = %s", session_id, host)
		self._cleanup_expired_sessions()

	def get_by_session(self, session_id, host):
		logging.debug("Looking up session %s" % session_id)

		res = self.db.get("SELECT uid FROM sessions WHERE session_id = %s \
			AND host = %s AND NOW() BETWEEN time_created AND time_expires",
			session_id, host)

		# Session does not exist or has expired
		if not res:
			return

		# Update the session expiration time
		self.db.execute("UPDATE sessions SET time_expires = NOW() + INTERVAL '14 days' \
			WHERE session_id = %s AND host = %s", session_id, host)

		return self.get_by_uid(res.uid)
		

class Account(Object):
	def __init__(self, backend, dn, attrs=None):
		Object.__init__(self, backend)
		self.dn = dn

		self.__attrs = attrs or {}

	def __str__(self):
		return self.name

	def __repr__(self):
		return "<%s %s>" % (self.__class__.__name__, self.dn)

	def __eq__(self, other):
		if isinstance(other, self.__class__):
			return self.dn == other.dn

	def __lt__(self, other):
		if isinstance(other, self.__class__):
			return self.name < other.name

	@property
	def ldap(self):
		return self.accounts.ldap

	@property
	def attributes(self):
		return self.__attrs

	def _get_first_attribute(self, attr, default=None):
		if attr not in self.attributes:
			return default

		res = self.attributes.get(attr, [])
		if res:
			return res[0]

	def get(self, key):
		try:
			attribute = self.attributes[key]
		except KeyError:
			raise AttributeError(key)

		if len(attribute) == 1:
			return attribute[0]

		return attribute

	def check_password(self, password):
		"""
			Bind to the server with given credentials and return
			true if password is corrent and false if not.

			Raises exceptions from the server on any other errors.
		"""

		logging.debug("Checking credentials for %s" % self.dn)
		try:
			self.ldap.simple_bind_s(self.dn, password.encode("utf-8"))
		except ldap.INVALID_CREDENTIALS:
			logging.debug("Account credentials are invalid.")
			return False

		logging.debug("Successfully authenticated.")
		return True

	def is_admin(self):
		return "wheel" in self.groups

	def is_talk_enabled(self):
		return "sipUser" in self.classes or "sipRoutingObject" in self.classes \
			or self.telephone_numbers or self.address

	@property
	def classes(self):
		return (x.decode() for x in self.attributes.get("objectClass", []))

	@property
	def uid(self):
		return self._get_first_attribute("uid").decode()

	@property
	def name(self):
		return self._get_first_attribute("cn").decode()

	@property
	def first_name(self):
		return self._get_first_attribute("givenName").decode()

	@lazy_property
	def groups(self):
		groups = []

		res = self.accounts._query("(&(objectClass=posixGroup) \
			(memberUid=%s))" % self.uid, ["cn"])

		for dn, attrs in res:
			cns = attrs.get("cn")
			if cns:
				groups.append(cns[0].decode())

		return groups

	@property
	def address(self):
		address = self._get_first_attribute("homePostalAddress", "".encode()).decode()
		address = address.replace(", ", "\n")

		return address

	@property
	def email(self):
		name = self.name.lower()
		name = name.replace(" ", ".")
		name = name.replace("Ä", "Ae")
		name = name.replace("Ö", "Oe")
		name = name.replace("Ü", "Ue")
		name = name.replace("ä", "ae")
		name = name.replace("ö", "oe")
		name = name.replace("ü", "ue")

		for mail in self.attributes.get("mail", []):
			if mail.decode().startswith("%s@ipfire.org" % name):
				return mail

		# If everything else fails, we will go with the UID
		return "%s@ipfire.org" % self.uid

	@property
	def sip_id(self):
		if "sipUser" in self.classes:
			return self._get_first_attribute("sipAuthenticationUser").decode()

		if "sipRoutingObject" in self.classes:
			return self._get_first_attribute("sipLocalAddress").decode()

	@property
	def sip_password(self):
		return self._get_first_attribute("sipPassword").decode()

	@property
	def sip_url(self):
		return "%s@ipfire.org" % self.sip_id

	def uses_sip_forwarding(self):
		if self.sip_routing_url:
			return True

		return False

	@property
	def sip_routing_url(self):
		if "sipRoutingObject" in self.classes:
			return self._get_first_attribute("sipRoutingAddress").decode()

	@lazy_property
	def sip_registrations(self):
		sip_registrations = []

		for reg in self.backend.talk.freeswitch.get_sip_registrations(self.sip_url):
			reg.account = self

			sip_registrations.append(reg)

		return sip_registrations

	@property
	def telephone_numbers(self):
		return self._telephone_numbers + self.mobile_telephone_numbers \
			+ self.home_telephone_numbers

	@property
	def _telephone_numbers(self):
		return self.attributes.get("telephoneNumber") or []

	@property
	def home_telephone_numbers(self):
		return self.attributes.get("homePhone") or []

	@property
	def mobile_telephone_numbers(self):
		return self.attributes.get("mobile") or []

	def avatar_url(self, size=None):
		if self.backend.debug:
			hostname = "http://accounts.dev.ipfire.org"
		else:
			hostname = "https://accounts.ipfire.org"

		url = "%s/avatar/%s.jpg" % (hostname, self.uid)

		if size:
			url += "?size=%s" % size

		return url

	def get_avatar(self, size=None):
		avatar = self._get_first_attribute("jpegPhoto")
		if not avatar:
			return

		if not size:
			return avatar

		return self._resize_avatar(avatar, size)

	def _resize_avatar(self, image, size):
		image = PIL.Image.open(io.BytesIO(image))

		# Convert RGBA images into RGB because JPEG doesn't support alpha-channels
		if image.mode == "RGBA":
			image = image.convert("RGB")

		# Resize the image to the desired resolution
		image.thumbnail((size, size), PIL.Image.ANTIALIAS)

		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, "JPEG", optimize=True, quality=98)
			except:
				image.save(f, "JPEG", quality=98)

			return f.getvalue()


if __name__ == "__main__":
	a = Accounts()

	print(a.list())