location: Remove blacklist feature

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

#!/usr/bin/python3

import hashlib
import hmac
import json
import logging
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 _sign_request(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()

	def _sign_response(self, body):
		h = hmac.new(self.api_secret.encode("utf-8"), body, hashlib.sha512)

		return h.hexdigest()

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

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

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

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

		# Log request
		logging.debug("Sending request to %s:" % request.url)
		for header in sorted(request.headers):
			logging.debug("	%s: %s" % (header, request.headers[header]))
		if request.body:
			logging.debug("%s" % json.dumps(kwargs, indent=4, sort_keys=True))

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

		# Log response
		logging.debug("Got response %s from %s in %.2fms:" % \
			(response.code, response.effective_url, response.request_time * 1000))
		for header in response.headers:
			logging.debug("	%s: %s" % (header, response.headers[header]))

		# Fetch the whole body
		body = response.body
		if body:
			# Decode the JSON response
			body = json.loads(body)

			# Log what we have received in a human-readable way
			logging.debug("%s" % json.dumps(body, indent=4, sort_keys=True))

		# Fetch the signature
		signature = response.headers.get("Hash")
		if not signature:
			raise RuntimeError("Could not find signature on response")

		expected_signature = self._sign_response(response.body)
		if not hmac.compare_digest(expected_signature, signature):
			raise RuntimeError("Invalid signature: %s" % signature)

		# Return the body
		return body