mirrors: Stop emailing Michael if a mirror is down

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

#!/usr/bin/python

import hashlib
import hmac
import json
import tornado.httpclient
import urllib.parse

from .misc import Object

class ZeiterfassungClient(Object):
	algorithm = "Zeiterfassung-HMAC-SHA512"

	def init(self):
		self.url = self.settings.get("zeiterfassung_url")

		# API credentials
		self.api_key    = self.settings.get("zeiterfassung_api_key")
		self.api_secret = self.settings.get("zeiterfassung_api_secret")

		# Check if all configuration values are set
		if not all((self.url, self.api_key, self.api_secret)):
			raise RuntimeError("%s is not configured" % self.__class__.__name__)

	def _make_signature(self, method, path, body):
		# Empty since we only support POST
		canonical_query = ""

		# Put everything together
		string_to_sign = "\n".join((
			method, path, canonical_query,
		)).encode("utf-8") + body

		# Compute HMAC
		h = hmac.new(self.api_secret.encode("utf-8"), string_to_sign, hashlib.sha512)

		return h.hexdigest()

	async def send_request(self, path, **kwargs):
		url = urllib.parse.urljoin(self.url, path)

		# Query arguments are all keyword arguments
		arguments = kwargs

		request = tornado.httpclient.HTTPRequest(url, method="POST")
		request.body = urllib.parse.urlencode(arguments)

		# Compose the signature
		signature = self._make_signature("POST", path, request.body)

		# Add authorization header
		request.headers["Authorization"] = " ".join(
			(self.algorithm, self.api_key, signature)
		)

		# Send the request
		response = await self.backend.http_client.fetch(request)

		# Decode the JSON response
		return json.loads(response.body.decode())