talk: Drop old FreeSWITCH integration

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

#!/usr/bin/python

import datetime
import dateutil.parser
import functools
import http.client
import ipaddress
import logging
import time
import tornado.locale
import tornado.web

from ..decorators import *
from .. import util

class ratelimit(object):
	def __init__(self, minutes=15, requests=180):
		self.minutes = minutes
		self.requests = requests

	def __call__(self, method):
		@functools.wraps(method)
		def wrapper(handler, *args, **kwargs):
			# Pass the request to the rate limiter and get a request object
			req = handler.backend.ratelimiter.handle_request(handler.request,
				handler, minutes=self.minutes, limit=self.requests)

			# If the rate limit has been reached, we won't allow
			# processing the request and therefore send HTTP error code 429.
			if req.is_ratelimited():
				raise tornado.web.HTTPError(429, "Rate limit exceeded")

			return method(handler, *args, **kwargs)

		return wrapper


class BaseHandler(tornado.web.RequestHandler):
	def prepare(self):
		# Mark this as private when someone is logged in
		if self.current_user:
			self.set_header("Cache-Control", "private")

		# Always send Vary: Cookie
		self.set_header("Vary", "Cookie")

	def set_expires(self, seconds):
		# For HTTP/1.1
		self.add_header("Cache-Control", "max-age=%s, must-revalidate" % seconds)

		# For HTTP/1.0
		expires = datetime.datetime.utcnow() + datetime.timedelta(seconds=seconds)
		self.set_header("Expires", expires)

	def write_error(self, status_code, **kwargs):
		# Translate code into message
		try:
			message = http.client.responses[status_code]
		except KeyError:
			message = None

		self.render("error.html", status_code=status_code, message=message, **kwargs)

	@property
	def hostname(self):
		# Return hostname in production
		if self.request.host.endswith("ipfire.org"):
			return self.request.host

		# Remove the development prefix
		subdomain, delimier, domain = self.request.host.partition(".")
		if subdomain:
			return "%s.ipfire.org" % subdomain

		# Return whatever it is
		return self.request.host

	def get_template_namespace(self):
		ns = tornado.web.RequestHandler.get_template_namespace(self)

		now = datetime.date.today()

		ns.update({
			"backend" : self.backend,
			"debug" : self.application.settings.get("debug", False),
			"format_size" : util.format_size,
			"format_time" : util.format_time,
			"hostname" : self.hostname,
			"now" : now,
			"year" : now.year,
		})

		return ns

	def get_remote_ip(self):
		# Fix for clients behind a proxy that sends "X-Forwarded-For".
		remote_ips = self.request.remote_ip.split(", ")

		for remote_ip in remote_ips:
			try:
				addr = ipaddress.ip_address(remote_ip)
			except ValueError:
				# Skip invalid IP addresses.
				continue

			# Check if the given IP address is from a
			# private network.
			if addr.is_private:
				continue

			return remote_ip

		# Return the last IP if nothing else worked
		return remote_ips.pop()

	@lazy_property
	def current_address(self):
		address = self.get_remote_ip()

		if address:
			return util.Address(self.backend, address)

	@lazy_property
	def current_country_code(self):
		if self.current_address:
			return self.current_address.country_code

	def get_argument_int(self, *args, **kwargs):
		arg = self.get_argument(*args, **kwargs)

		if arg is None or arg == "":
			return

		try:
			return int(arg)
		except ValueError:
			raise tornado.web.HTTPError(400, "Could not convert integer: %s" % arg)

	def get_argument_float(self, *args, **kwargs):
		arg = self.get_argument(*args, **kwargs)

		if arg is None or arg == "":
			return

		try:
			return float(arg)
		except ValueError:
			raise tornado.web.HTTPError(400, "Could not convert float: %s" % arg)

	def get_argument_date(self, arg, *args, **kwargs):
		value = self.get_argument(arg, *args, **kwargs)
		if value is None:
			return

		try:
			return dateutil.parser.parse(value)
		except ValueError:
			raise tornado.web.HTTPError(400)

	def get_file(self, name):
		try:
			file = self.request.files[name][0]

			return file["filename"], file["body"], file["content_type"]
		except KeyError:
			return None

	# Login stuff

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

		# Get account from the session object
		account = self.backend.accounts.get_by_session(session_id, self.request.host)

		# If the account was not found or the session was not valid
		# any more, we will remove the cookie.
		if not account:
			self.clear_cookie("session_id")

		return account

	@property
	def backend(self):
		return self.application.backend

	@property
	def db(self):
		return self.backend.db

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

	@property
	def downloads(self):
		return self.backend.downloads

	@property
	def fireinfo(self):
		return self.backend.fireinfo

	@property
	def iuse(self):
		return self.backend.iuse

	@property
	def memcached(self):
		return self.backend.memcache

	@property
	def mirrors(self):
		return self.backend.mirrors

	@property
	def netboot(self):
		return self.backend.netboot

	@property
	def releases(self):
		return self.backend.releases


class APIHandler(BaseHandler):
	def check_xsrf_cookie(self):
		"""
			Do nothing here, because we cannot verify the XSRF token
		"""
		pass


class NotFoundHandler(BaseHandler):
	def prepare(self):
		# Raises 404 as soon as it is called
		raise tornado.web.HTTPError(404)


class ErrorHandler(BaseHandler):
	"""
		Raises any error we want
	"""
	def get(self, code):
		try:
			code = int(code)
		except:
			raise tornado.web.HTTPError(400)

		raise tornado.web.HTTPError(code)