Migrate to libloc

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

#!/usr/bin/python

import datetime
import logging
import iso3166
import math
import os.path
import random
import socket
import time
import ssl
import tornado.httpclient
import tornado.iostream
import tornado.netutil
import urllib.parse

from . import countries
from . import util
from .misc import Object
from .decorators import *

class Mirrors(Object):
	def _get_mirrors(self, query, *args):
		res = self.db.query(query, *args)

		for row in res:
			yield Mirror(self.backend, row.id, data=row)

	def _get_mirror(self, query, *args):
		res = self.db.get(query, *args)

		if res:
			return Mirror(self.backend, res.id, data=res)

	def __iter__(self):
		mirrors = self._get_mirrors("SELECT * FROM mirrors \
			WHERE enabled IS TRUE ORDER BY hostname")

		return iter(mirrors)

	async def check_all(self):
		for mirror in self:
			with self.db.transaction():
				await mirror.check()

	def get_for_download(self, filename, country_code=None):
		# Try to find a good mirror for this country first
		if country_code:
			zone = countries.get_zone(country_code)

			mirror = self._get_mirror("SELECT mirrors.* FROM mirror_files files \
				LEFT JOIN mirrors ON files.mirror = mirrors.id \
				WHERE files.filename = %s \
					AND mirrors.enabled IS TRUE AND mirrors.state = %s \
					AND mirrors.country_code = ANY(%s) \
					ORDER BY RANDOM() LIMIT 1", filename, "UP",
					countries.get_in_zone(zone))

			if mirror:
				return mirror

		# Get a random mirror that serves the file
		return self._get_mirror("SELECT mirrors.* FROM mirror_files files \
			LEFT JOIN mirrors ON files.mirror = mirrors.id \
			WHERE files.filename = %s \
				AND mirrors.enabled IS TRUE AND mirrors.state = %s \
				ORDER BY RANDOM() LIMIT 1", filename, "UP")

	def get_by_hostname(self, hostname):
		return self._get_mirror("SELECT * FROM mirrors \
			WHERE hostname = %s", hostname)

	def get_by_countries(self):
		mirrors = {}

		for m in self:
			try:
				mirrors[m.country].append(m)
			except KeyError:
				mirrors[m.country] = [m]

		return mirrors


class Mirror(Object):
	def init(self, id, data=None):
		self.id   = id
		self.data = data

	def __str__(self):
		return self.hostname

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

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

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

	def __hash__(self):
		return self.id

	@lazy_property
	def url(self):
		url = "%s://%s" % ("https" if self.supports_https else "http", self.hostname)

		if not self.path.startswith("/"):
			url += "/"

		url += "%s" % self.path

		if not self.path.endswith("/"):
			url += "/"

		return url

	@property
	def hostname(self):
		return self.data.hostname

	@lazy_property
	def address(self):
		"""
			Returns the stored address
		"""
		if self.data.address:
			return util.Address(self.backend, self.data.address)

	@property
	def path(self):
		return self.data.path

	@property
	def supports_https(self):
		return self.data.supports_https

	@property
	def owner(self):
		return self.data.owner

	@property
	def country(self):
		return iso3166.countries.get(self.country_code)

	@property
	def country_code(self):
		if self.data.country_code:
			return self.data.country_code

		if self.address:
			return self.address.country_code

	@property
	def zone(self):
		return countries.get_zone(self.country_name)

	@property
	def asn(self):
		if self.address:
			return self.address.asn

	@property
	def filelist(self):
		filelist = self.db.query("SELECT filename FROM mirror_files WHERE mirror=%s ORDER BY filename", self.id)
		return [f.filename for f in filelist]

	@property
	def prefix(self):
		return ""

	def build_url(self, filename):
		return urllib.parse.urljoin(self.url, filename)

	@property
	def last_update(self):
		return self.data.last_update

	@property
	def state(self):
		return self.data.state

	def set_state(self, state):
		logging.debug("Setting state of %s to %s" % (self.hostname, state))

		if self.state == state:
			return

		self.db.execute("UPDATE mirrors SET state = %s WHERE id = %s", state, self.id)

		# Reload changed settings
		self.data["state"] = state

	@property
	def enabled(self):
		return self.data.enabled

	@property
	def disabled(self):
		return not self.enabled

	async def check(self):
		logging.debug("Running check for mirror %s" % self.hostname)

		self.db.execute("UPDATE mirrors SET address = %s WHERE id = %s",
			await self.resolve(), self.id)

		success = await self.check_timestamp()
		if success:
			await self.check_filelist()

	def check_state(self, last_update):
		logging.debug("Checking state of mirror %s" % self.id)

		if not self.enabled:
			self.set_state("DOWN")
			return

		now = datetime.datetime.utcnow()

		time_delta = now - last_update
		time_diff = time_delta.total_seconds()

		time_down = self.settings.get_int("mirrors_time_down", 3*24*60*60)
		if time_diff >= time_down:
			self.set_state("DOWN")
			return

		time_outofsync = self.settings.get_int("mirrors_time_outofsync", 6*60*60)
		if time_diff >= time_outofsync:
			self.set_state("OUTOFSYNC")
			return

		self.set_state("UP")

	async def check_timestamp(self):
		http = tornado.httpclient.AsyncHTTPClient()

		try:
			response = await http.fetch(self.url + ".timestamp",
				headers={ "Pragma" : "no-cache" })
		except tornado.httpclient.HTTPError as e:
			logging.warning("Error getting timestamp from %s: %s" % (self.hostname, e))
			self.set_state("DOWN")
			return False

		except ssl.SSLError as e:
			logging.warning("SSL error when getting timestamp from %s: %s" % (self.hostname, e))
			self.set_state("DOWN")
			return False

		except tornado.iostream.StreamClosedError as e:
			logging.warning("Connection closed unexpectedly for %s: %s" % (self.hostname, e))
			self.set_state("DOWN")
			return False

		except OSError as e:
			logging.warning("Could not connect to %s: %s" % (self.hostname, e))
			self.set_state("DOWN")
			return False

		if response.error:
			logging.warning("Error getting timestamp from %s" % self.hostname)
			self.set_state("DOWN")
			return

		try:
			timestamp = int(response.body.strip())
		except ValueError:
			timestamp = 0

		timestamp = datetime.datetime.utcfromtimestamp(timestamp)

		self.db.execute("UPDATE mirrors SET last_update = %s WHERE id = %s",
			timestamp, self.id)

		# Update state
		self.check_state(timestamp)

		logging.debug("Successfully updated timestamp from %s" % self.hostname)

		return True

	async def check_filelist(self):
		# XXX need to remove data from disabled mirrors
		if not self.enabled:
			return

		http = tornado.httpclient.AsyncHTTPClient()

		try:
			response = await http.fetch(self.url + ".filelist",
				headers={ "Pragma" : "no-cache" })
		except tornado.httpclient.HTTPError as e:
			logging.warning("Error getting filelist from %s: %s" % (self.hostname, e))
			self.set_state("DOWN")
			return

		if response.error:
			logging.debug("Error getting filelist from %s" % self.hostname)
			return

		# Drop the old filelist
		self.db.execute("DELETE FROM mirror_files WHERE mirror = %s", self.id)

		# Add them all again
		for file in response.body.splitlines():
			file = os.path.join(self.prefix, file.decode())

			self.db.execute("INSERT INTO mirror_files(mirror, filename) \
				VALUES(%s, %s)", self.id, file)

		logging.debug("Successfully updated mirror filelist from %s" % self.hostname)

	@property
	def development(self):
		return self.data.get("mirrorlist_devel", False)

	@property
	def mirrorlist(self):
		return self.data.get("mirrorlist", False)

	async def resolve(self):
		"""
			Returns a single IP address of this mirror
		"""
		addresses = await self.backend.resolver.resolve(self.hostname, 0)

		# Return the first address
		for family, address in addresses:
			host, port = address

			return host