[people/stevee/pakfire.git] / python / pakfire / transport.py

#!/usr/bin/python
###############################################################################
#                                                                             #
# Pakfire - The IPFire package management system                              #
# Copyright (C) 2013 Pakfire development team                                 #
#                                                                             #
# This program is free software: you can redistribute it and/or modify        #
# it under the terms of the GNU General Public License as published by        #
# the Free Software Foundation, either version 3 of the License, or           #
# (at your option) any later version.                                         #
#                                                                             #
# This program is distributed in the hope that it will be useful,             #
# but WITHOUT ANY WARRANTY; without even the implied warranty of              #
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the               #
# GNU General Public License for more details.                                #
#                                                                             #
# You should have received a copy of the GNU General Public License           #
# along with this program.  If not, see <http://www.gnu.org/licenses/>.       #
#                                                                             #
###############################################################################

from __future__ import division

import base64
import hashlib
import json
import os
import time
import urlgrabber
import urllib
import urlparse

import pakfire.downloader
import pakfire.util

from pakfire.constants import *
from pakfire.i18n import _

import logging
log = logging.getLogger("pakfire.transport")


class PakfireHubTransportUploader(object):
	"""
		Handles the upload of a single file to the hub.
	"""

	def __init__(self, transport, filename):
		self.transport = transport
		self.filename = filename

	def get_upload_id(self):
		"""
			Gets an upload from the pakfire hub.
		"""
		# Calculate the SHA1 sum of the file to upload.
		h = hashlib.new("sha1")
		with open(self.filename, "rb") as f:
			while True:
				buf = f.read(CHUNK_SIZE)
				if not buf:
					break

				h.update(buf)

		data = {
			"filename" : os.path.basename(self.filename),
			"filesize" : os.path.getsize(self.filename),
			"hash"     : h.hexdigest(),
		}

		upload_id = self.transport.get("/uploads/create", data=data)
		log.debug("Got upload id: %s" % upload_id)

		return upload_id

	def send_file(self, upload_id, progress_callback=None):
		"""
			Sends the file content to the server.

			The data is splitted into chunks, which are
			sent one after an other.
		"""
		with open(self.filename, "rb") as f:
			# Initial chunk size.
			chunk_size = CHUNK_SIZE

			# Count the already transmitted bytes.
			transferred = 0

			while True:
				chunk = f.read(chunk_size)
				if not chunk:
					break

				log.debug("Got chunk of %s bytes" % len(chunk))

				# Save the time when we started to send this bit.
				time_started = time.time()

				# Send the chunk to the server.
				self.send_chunk(upload_id, chunk)

				# Save the duration.time after the chunk has been transmitted
				# and adjust chunk size to send one chunk per second.
				duration = time.time() - time_started
				chunk_size = int(chunk_size / duration)

				# Never let chunk_size drop under CHUNK_SIZE:
				if chunk_size < CHUNK_SIZE:
					chunk_size = CHUNK_SIZE

				# Add up the send amount of data.
				transferred += len(chunk)
				if progress_callback:
					progress_callback(transferred)

	def send_chunk(self, upload_id, data):
		"""
			Sends a piece of the file to the server.
		"""
		# Calculate checksum over the chunk data.
		h = hashlib.new("sha512")
		h.update(data)
		chksum = h.hexdigest()

		# Encode data in base64.
		data = base64.b64encode(data)

		# Send chunk data to the server.
		self.transport.post("/uploads/%s/sendchunk" % upload_id,
			data={ "chksum" : chksum, "data" : data })

	def destroy_upload(self, upload_id):
		"""
			Destroys the upload on the server.
		"""
		self.transport.get("/uploads/%s/destroy" % upload_id)

	def finish_upload(self, upload_id):
		"""
			Signals to the server, that the upload has finished.
		"""
		self.transport.get("/uploads/%s/finished" % upload_id)

	def run(self):
		upload_id = None

		# Create a progress bar.
		progress = pakfire.util.make_progress(
			os.path.basename(self.filename), os.path.getsize(self.filename), speed=True, eta=True,
		)

		try:
			# Get an upload ID.
			upload_id = self.get_upload_id()

			# Send the file content.
			if progress:
				self.send_file(upload_id, progress_callback=progress.update)
			else:
				self.send_file(upload_id)

		except:
			if progress:
				progress.finish()

			# Remove broken upload from server.
			if upload_id:
				self.destroy_upload(upload_id)

			# XXX catch fatal errors
			raise

		else:
			if progress:
				progress.finish()

			# If no exception was raised, the upload
			# has finished.
			self.finish_upload(upload_id)

		# Return the upload id so some code can actually do something
		# with the file on the server.
		return upload_id


class PakfireHubTransport(object):
	"""
		Connection to the pakfire hub.
	"""

	def __init__(self, config):
		self.config = config

		# Create connection to the hub.
		self.grabber = pakfire.downloader.PakfireGrabber(
			self.config, prefix=self.url,
		)

	@property
	def url(self):
		"""
			Construct a right URL out of the given
			server, username and password.

			Basicly this just adds the credentials
			to the URL.
		"""
		# Get credentials.
		server, username, password = self.config.get_hub_credentials()

		# Parse the given URL.
		url = urlparse.urlparse(server)
		assert url.scheme in ("http", "https")

		# Build new URL.
		ret = "%s://" % url.scheme

		# Add credentials if provided.
		if username and password:
			ret += "%s:%s@" % (username, password)

		# Add path components.
		ret += url.netloc

		return ret

	def one_request(self, url, **kwargs):
		try:
			return self.grabber.urlread(url, **kwargs)

		except urlgrabber.grabber.URLGrabError, e:
			# Timeout
			if e.errno == 12:
				raise TransportConnectionTimeoutError, e

			# Handle common HTTP errors
			elif e.errno == 14:
				# Connection errors
				if e.code == 5:
					raise TransportConnectionProxyError, url
				elif e.code == 6:
					raise TransportConnectionDNSError, url
				elif e.code == 7:
					raise TransportConnectionResetError, url
				elif e.code == 23:
					raise TransportConnectionWriteError, url
				elif e.code == 26:
					raise TransportConnectionReadError, url

				# SSL errors
				elif e.code == 52:
					raise TransportSSLCertificateExpiredError, url

				# HTTP error codes
				elif e.code == 403:
					raise TransportForbiddenError, url
				elif e.code == 404:
					raise TransportNotFoundError, url
				elif e.code == 500:
					raise TransportInternalServerError, url
				elif e.code == 504:
					raise TransportConnectionTimeoutError, url

			# All other exceptions...
			raise

	def request(self, url, tries=None, **kwargs):
		# tries = None implies wait infinitely

		while tries or tries is None:
			if tries:
				tries -= 1

			try:
				return self.one_request(url, **kwargs)

			# 500 - Internal Server Error
			except TransportInternalServerError, e:
				log.exception("%s" % e.__class__.__name__)

				# Wait a minute before trying again.
				time.sleep(60)

			# Retry on connection problems.
			except TransportConnectionError, e:
				log.exception("%s" % e.__class__.__name__)

				# Wait for 10 seconds.
				time.sleep(10)

		raise TransportMaxTriesExceededError

	def escape_args(self, **kwargs):
		return urllib.urlencode(kwargs)

	def get(self, url, data={}, **kwargs):
		"""
			Sends a HTTP GET request to the given URL.

			All given keyword arguments are considered as form data.
		"""
		params = self.escape_args(**data)

		if params:
			url = "%s?%s" % (url, params)

		return self.request(url, **kwargs)

	def post(self, url, data={}, **kwargs):
		"""
			Sends a HTTP POST request to the given URL.

			All keyword arguments are considered as form data.
		"""
		params = self.escape_args(**data)
		if params:
			kwargs.update({
				"data" : params,
			})

		return self.request(url, **kwargs)

	def upload_file(self, filename):
		"""
			Uploads the given file to the server.
		"""
		uploader = PakfireHubTransportUploader(self, filename)
		upload_id = uploader.run()

		return upload_id

	def get_json(self, *args, **kwargs):
		res = self.get(*args, **kwargs)

		# Decode JSON.
		if res:
			return json.loads(res)

	### Misc. actions

	def noop(self):
		"""
			No operation. Just to check if the connection is
			working. Returns a random number.
		"""
		return self.get("/noop")

	def test_code(self, error_code):
		assert error_code >= 100 and error_code <= 999

		self.get("/error/test/%s" % error_code)

	# Build actions

	def build_create(self, filename, build_type, arches=None, distro=None):
		"""
			Create a new build on the hub.
		"""
		assert build_type in ("scratch", "release")

		# XXX Check for permission to actually create a build.

		# Upload the source file to the server.
		upload_id = self.upload_file(filename)

		data = {
			"arches"     : ",".join(arches or []),
			"build_type" : build_type,
			"distro"     : distro or "",
			"upload_id"  : upload_id,
		}

		# Then create the build.
		build_id = self.get("/builds/create", data=data)

		return build_id or None

	def build_get(self, build_uuid):
		return self.get_json("/builds/%s" % build_uuid)

	# Job actions

	def job_get(self, job_uuid):
		return self.get_json("/jobs/%s" % job_uuid)

	# Package actions

	def package_get(self, package_uuid):
		return self.get_json("/packages/%s" % package_uuid)
Commit	Line	Data
aa14071d MT	1	#!/usr/bin/python
	2	###############################################################################
	3	# #
	4	# Pakfire - The IPFire package management system #
	5	# Copyright (C) 2013 Pakfire development team #
	6	# #
	7	# This program is free software: you can redistribute it and/or modify #
	8	# it under the terms of the GNU General Public License as published by #
	9	# the Free Software Foundation, either version 3 of the License, or #
	10	# (at your option) any later version. #
	11	# #
	12	# This program is distributed in the hope that it will be useful, #
	13	# but WITHOUT ANY WARRANTY; without even the implied warranty of #
	14	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the #
	15	# GNU General Public License for more details. #
	16	# #
	17	# You should have received a copy of the GNU General Public License #
	18	# along with this program. If not, see <http://www.gnu.org/licenses/>. #
	19	# #
	20	###############################################################################
	21
	22	from __future__ import division
	23
	24	import base64
	25	import hashlib
	26	import json
	27	import os
	28	import time
	29	import urlgrabber
	30	import urllib
	31	import urlparse
	32
	33	import pakfire.downloader
	34	import pakfire.util
	35
	36	from pakfire.constants import *
	37	from pakfire.i18n import _
	38
	39	import logging
	40	log = logging.getLogger("pakfire.transport")
	41
	42
	43	class PakfireHubTransportUploader(object):
	44	"""
	45	Handles the upload of a single file to the hub.
	46	"""
	47
	48	def __init__(self, transport, filename):
	49	self.transport = transport
	50	self.filename = filename
	51
	52	def get_upload_id(self):
	53	"""
	54	Gets an upload from the pakfire hub.
	55	"""
	56	# Calculate the SHA1 sum of the file to upload.
	57	h = hashlib.new("sha1")
	58	with open(self.filename, "rb") as f:
	59	while True:
	60	buf = f.read(CHUNK_SIZE)
	61	if not buf:
	62	break
	63
	64	h.update(buf)
65
66	data = {
67	"filename" : os.path.basename(self.filename),
68	"filesize" : os.path.getsize(self.filename),
69	"hash" : h.hexdigest(),
70	}
71
72	upload_id = self.transport.get("/uploads/create", data=data)
73	log.debug("Got upload id: %s" % upload_id)
74
75	return upload_id
76
77	def send_file(self, upload_id, progress_callback=None):
78	"""
79	Sends the file content to the server.
80
81	The data is splitted into chunks, which are
82	sent one after an other.
83	"""
84	with open(self.filename, "rb") as f:
85	# Initial chunk size.
86	chunk_size = CHUNK_SIZE
87
88	# Count the already transmitted bytes.
89	transferred = 0
90
91	while True:
92	chunk = f.read(chunk_size)
93	if not chunk:
94	break
95
96	log.debug("Got chunk of %s bytes" % len(chunk))
97
98	# Save the time when we started to send this bit.
99	time_started = time.time()
100
101	# Send the chunk to the server.
102	self.send_chunk(upload_id, chunk)
103
104	# Save the duration.time after the chunk has been transmitted
105	# and adjust chunk size to send one chunk per second.
106	duration = time.time() - time_started
107	chunk_size = int(chunk_size / duration)
108
109	# Never let chunk_size drop under CHUNK_SIZE:
110	if chunk_size < CHUNK_SIZE:
111	chunk_size = CHUNK_SIZE
112
113	# Add up the send amount of data.
114	transferred += len(chunk)
115	if progress_callback:
116	progress_callback(transferred)
117
118	def send_chunk(self, upload_id, data):
119	"""
120	Sends a piece of the file to the server.
121	"""
122	# Calculate checksum over the chunk data.
123	h = hashlib.new("sha512")
124	h.update(data)
125	chksum = h.hexdigest()
126
127	# Encode data in base64.
128	data = base64.b64encode(data)
129
130	# Send chunk data to the server.
131	self.transport.post("/uploads/%s/sendchunk" % upload_id,
132	data={ "chksum" : chksum, "data" : data })
133
134	def destroy_upload(self, upload_id):
135	"""
136	Destroys the upload on the server.
137	"""
138	self.transport.get("/uploads/%s/destroy" % upload_id)
139
140	def finish_upload(self, upload_id):
141	"""
142	Signals to the server, that the upload has finished.
143	"""
144	self.transport.get("/uploads/%s/finished" % upload_id)
145
146	def run(self):
147	upload_id = None
148
149	# Create a progress bar.
150	progress = pakfire.util.make_progress(
151	os.path.basename(self.filename), os.path.getsize(self.filename), speed=True, eta=True,
152	)
153
154	try:
155	# Get an upload ID.
156	upload_id = self.get_upload_id()
157
158	# Send the file content.
17f2125d MT	159	if progress:
	160	self.send_file(upload_id, progress_callback=progress.update)
	161	else:
	162	self.send_file(upload_id)
aa14071d MT	163
aa14071d MT	164	except:
92470f48 MT	165	if progress:
92470f48 MT	166	progress.finish()
aa14071d MT	167
	168	# Remove broken upload from server.
	169	if upload_id:
	170	self.destroy_upload(upload_id)
	171
	172	# XXX catch fatal errors
	173	raise
	174
	175	else:
92470f48 MT	176	if progress:
92470f48 MT	177	progress.finish()
aa14071d MT	178
	179	# If no exception was raised, the upload
	180	# has finished.
	181	self.finish_upload(upload_id)
	182
	183	# Return the upload id so some code can actually do something
	184	# with the file on the server.
	185	return upload_id
	186
	187
	188	class PakfireHubTransport(object):
	189	"""
	190	Connection to the pakfire hub.
	191	"""
	192
	193	def __init__(self, config):
	194	self.config = config
	195
	196	# Create connection to the hub.
	197	self.grabber = pakfire.downloader.PakfireGrabber(
	198	self.config, prefix=self.url,
	199	)
	200
aa14071d MT	201	@property
	202	def url(self):
	203	"""
	204	Construct a right URL out of the given
	205	server, username and password.
	206
	207	Basicly this just adds the credentials
	208	to the URL.
	209	"""
	210	# Get credentials.
	211	server, username, password = self.config.get_hub_credentials()
	212
	213	# Parse the given URL.
	214	url = urlparse.urlparse(server)
	215	assert url.scheme in ("http", "https")
	216
	217	# Build new URL.
	218	ret = "%s://" % url.scheme
	219
	220	# Add credentials if provided.
	221	if username and password:
	222	ret += "%s:%s@" % (username, password)
	223
	224	# Add path components.
	225	ret += url.netloc
	226
	227	return ret
	228
	229	def one_request(self, url, **kwargs):
	230	try:
	231	return self.grabber.urlread(url, **kwargs)
	232
	233	except urlgrabber.grabber.URLGrabError, e:
	234	# Timeout
	235	if e.errno == 12:
	236	raise TransportConnectionTimeoutError, e
	237
	238	# Handle common HTTP errors
	239	elif e.errno == 14:
	240	# Connection errors
	241	if e.code == 5:
	242	raise TransportConnectionProxyError, url
	243	elif e.code == 6:
	244	raise TransportConnectionDNSError, url
	245	elif e.code == 7:
	246	raise TransportConnectionResetError, url
	247	elif e.code == 23:
	248	raise TransportConnectionWriteError, url
	249	elif e.code == 26:
	250	raise TransportConnectionReadError, url
	251
	252	# SSL errors
	253	elif e.code == 52:
	254	raise TransportSSLCertificateExpiredError, url
	255
	256	# HTTP error codes
	257	elif e.code == 403:
	258	raise TransportForbiddenError, url
	259	elif e.code == 404:
	260	raise TransportNotFoundError, url
	261	elif e.code == 500:
	262	raise TransportInternalServerError, url
6a05651d MT	263	elif e.code == 504:
6a05651d MT	264	raise TransportConnectionTimeoutError, url
aa14071d MT	265
	266	# All other exceptions...
	267	raise
	268
	269	def request(self, url, tries=None, **kwargs):
	270	# tries = None implies wait infinitely
	271
	272	while tries or tries is None:
	273	if tries:
	274	tries -= 1
	275
	276	try:
	277	return self.one_request(url, **kwargs)
	278
	279	# 500 - Internal Server Error
	280	except TransportInternalServerError, e:
	281	log.exception("%s" % e.__class__.__name__)
	282
	283	# Wait a minute before trying again.
	284	time.sleep(60)
	285
	286	# Retry on connection problems.
	287	except TransportConnectionError, e:
	288	log.exception("%s" % e.__class__.__name__)
	289
	290	# Wait for 10 seconds.
	291	time.sleep(10)
	292
	293	raise TransportMaxTriesExceededError
	294
	295	def escape_args(self, **kwargs):
	296	return urllib.urlencode(kwargs)
	297
	298	def get(self, url, data={}, **kwargs):
	299	"""
	300	Sends a HTTP GET request to the given URL.
	301
	302	All given keyword arguments are considered as form data.
	303	"""
	304	params = self.escape_args(**data)
	305
	306	if params:
	307	url = "%s?%s" % (url, params)
	308
	309	return self.request(url, **kwargs)
	310
	311	def post(self, url, data={}, **kwargs):
	312	"""
	313	Sends a HTTP POST request to the given URL.
	314
	315	All keyword arguments are considered as form data.
	316	"""
	317	params = self.escape_args(**data)
	318	if params:
	319	kwargs.update({
	320	"data" : params,
	321	})
	322
	323	return self.request(url, **kwargs)
	324
	325	def upload_file(self, filename):
	326	"""
	327	Uploads the given file to the server.
	328	"""
329	uploader = PakfireHubTransportUploader(self, filename)
330	upload_id = uploader.run()
331
332	return upload_id
333
334	def get_json(self, args, *kwargs):
335	res = self.get(args, *kwargs)
336
337	# Decode JSON.
338	if res:
339	return json.loads(res)
340
341	### Misc. actions
342
343	def noop(self):
344	"""
345	No operation. Just to check if the connection is
346	working. Returns a random number.
347	"""
348	return self.get("/noop")
349
350	def test_code(self, error_code):
351	assert error_code >= 100 and error_code <= 999
352
353	self.get("/error/test/%s" % error_code)
354
355	# Build actions
356
357	def build_create(self, filename, build_type, arches=None, distro=None):
358	"""
359	Create a new build on the hub.
360	"""
361	assert build_type in ("scratch", "release")
362
363	# XXX Check for permission to actually create a build.
364
365	# Upload the source file to the server.
366	upload_id = self.upload_file(filename)
367
368	data = {
369	"arches" : ",".join(arches or []),
370	"build_type" : build_type,
371	"distro" : distro or "",
372	"upload_id" : upload_id,
373	}
374
375	# Then create the build.
376	build_id = self.get("/builds/create", data=data)
377
378	return build_id or None
379
380	def build_get(self, build_uuid):
381	return self.get_json("/builds/%s" % build_uuid)
382
383	# Job actions
384
385	def job_get(self, job_uuid):
386	return self.get_json("/jobs/%s" % job_uuid)
387
388	# Package actions
389
390	def package_get(self, package_uuid):
391	return self.get_json("/packages/%s" % package_uuid)