SocialPub/tools/pasture/town/core/http.py

"""HTTP to the pasture: every https://<name>.test site through Caddy on 127.0.0.1:6443, verified against Caddy's own CA
(.ca/root.crt), with one kept-alive connection per site and thread. 429s wait for the server's reset; connection drops
are retried. Every request is timed into `Client.timings` for the load mode."""
import http.client
import json as jsonlib
import os
import socket
import ssl
import threading
import time
import urllib.parse
import uuid

CADDY = ("127.0.0.1", 6443)


class HttpError(Exception):
	def __init__(self, method, url, status, body):
		super().__init__(f"{method} {url} -> {status}: {body[:300]!r}")
		self.status = status
		self.body = body


class Response:
	def __init__(self, status, headers, body, url):
		self.status = status
		self.headers = headers
		self.body = body
		self.url = url

	@property
	def ok(self):
		return 200 <= self.status < 300

	@property
	def text(self):
		return self.body.decode("utf-8", "replace")

	def json(self):
		if not self.body:
			return None
		try:
			return jsonlib.loads(self.body)
		except ValueError:
			return None

	def header(self, name):
		for k, v in self.headers:
			if k.lower() == name.lower():
				return v
		return None


class _CaddyConnection(http.client.HTTPSConnection):
	"""Connects to Caddy's port but speaks TLS (SNI and certificate check) as the named site."""

	def __init__(self, site, context, timeout):
		super().__init__(site, 443, context=context, timeout=timeout)
		self._site = site
		self._ctx = context

	def connect(self):
		sock = socket.create_connection(CADDY, self.timeout)
		self.sock = self._ctx.wrap_socket(sock, server_hostname=self._site)


class Client:
	def __init__(self, ca_file, timeout=60):
		self.ctx = ssl.create_default_context(cafile=ca_file)
		self.timeout = timeout
		self.local = threading.local()
		self.timings = []  # (method, site, path template, status, ms)
		self.lock = threading.Lock()
		self.record = False

	def _connection(self, scheme, netloc):
		conns = getattr(self.local, "conns", None)
		if conns is None:
			conns = self.local.conns = {}
		key = (scheme, netloc)
		conn = conns.get(key)
		if conn is None:
			if scheme == "https":
				conn = _CaddyConnection(netloc.split(":")[0], self.ctx, self.timeout)
			else:
				host, _, port = netloc.partition(":")
				conn = http.client.HTTPConnection(host, int(port or 80), timeout=self.timeout)
			conns[key] = conn
		return conn

	def _drop(self, scheme, netloc):
		conn = self.local.conns.pop((scheme, netloc), None)
		if conn is not None:
			conn.close()

	def request(self, method, url, *, headers=None, json=None, form=None, data=None, files=None, params=None,
				ok=None, attempts=4, wait_429=True, template=None):
		"""Sends one request. `json` is a JSON body, `form` urlencoded pairs (a dict or a list of pairs, lists expand to
		key[]=v as Rails expects), `files` a multipart body ({name: (filename, bytes, type)} plus `form` fields).
		`ok` is a set of accepted statuses: anything else raises HttpError."""
		parts = urllib.parse.urlsplit(url)
		path = parts.path or "/"
		query = parts.query
		if params:
			extra = urllib.parse.urlencode(_pairs(params), doseq=True)
			query = f"{query}&{extra}" if query else extra
		target = path + (f"?{query}" if query else "")
		hdrs = {"Accept": "application/json", "User-Agent": "pasture-town/1"}
		body = None
		if json is not None:
			body = jsonlib.dumps(json).encode()
			hdrs["Content-Type"] = "application/json"
		elif files is not None:
			boundary = uuid.uuid4().hex
			body = _multipart(boundary, _pairs(form or {}), files)
			hdrs["Content-Type"] = f"multipart/form-data; boundary={boundary}"
		elif form is not None:
			body = urllib.parse.urlencode(_pairs(form), doseq=True).encode()
			hdrs["Content-Type"] = "application/x-www-form-urlencoded"
		elif data is not None:
			body = data
		if headers:
			hdrs.update(headers)
		last = None
		for attempt in range(attempts):
			conn = self._connection(parts.scheme, parts.netloc)
			started = time.monotonic()
			try:
				conn.request(method, target, body=body, headers=hdrs)
				raw = conn.getresponse()
				payload = raw.read()
				resp = Response(raw.status, raw.getheaders(), payload, url)
			except (ConnectionError, http.client.HTTPException, socket.timeout, ssl.SSLError, OSError) as e:
				self._drop(parts.scheme, parts.netloc)
				last = e
				time.sleep(0.5 * (attempt + 1))
				continue
			ms = (time.monotonic() - started) * 1000
			if self.record:
				with self.lock:
					self.timings.append((method, parts.netloc, template or path, resp.status, ms))
			if resp.status == 429 and wait_429 and attempt < attempts - 1:
				time.sleep(_retry_after(resp))
				continue
			if resp.status in (502, 503, 504) and attempt < attempts - 1:
				time.sleep(1 + attempt)
				continue
			if ok is not None and resp.status not in ok:
				raise HttpError(method, url, resp.status, resp.text)
			return resp
		raise HttpError(method, url, 0, repr(last))

	def get(self, url, **kw):
		return self.request("GET", url, **kw)

	def post(self, url, **kw):
		return self.request("POST", url, **kw)


def _pairs(values):
	"""A dict or list of pairs, with list values as Rails' key[]=v (unless the key already ends in [])."""
	items = values.items() if isinstance(values, dict) else values
	out = []
	for k, v in items:
		if v is None:
			continue
		if isinstance(v, (list, tuple)):
			key = k if k.endswith("[]") else f"{k}[]"
			out.extend((key, _scalar(x)) for x in v)
		else:
			out.append((k, _scalar(v)))
	return out


def _scalar(v):
	if v is True:
		return "true"
	if v is False:
		return "false"
	return str(v)


def _multipart(boundary, fields, files):
	chunks = []
	for k, v in fields:
		chunks.append(f'--{boundary}\r\nContent-Disposition: form-data; name="{k}"\r\n\r\n{v}\r\n'.encode())
	for name, (filename, content, ctype) in files.items():
		chunks.append(f'--{boundary}\r\nContent-Disposition: form-data; name="{name}"; filename="{filename}"\r\n'
					  f"Content-Type: {ctype}\r\n\r\n".encode() + content + b"\r\n")
	chunks.append(f"--{boundary}--\r\n".encode())
	return b"".join(chunks)


def _retry_after(resp):
	value = resp.header("Retry-After")
	if value and value.isdigit():
		return min(int(value), 60)
	reset = resp.header("X-RateLimit-Reset")
	if reset:
		try:
			from email.utils import parsedate_to_datetime
			import datetime
			when = datetime.datetime.fromisoformat(reset.replace("Z", "+00:00")) if "T" in reset \
				else parsedate_to_datetime(reset)
			return max(1, min(60, (when - datetime.datetime.now(datetime.timezone.utc)).total_seconds()))
		except (ValueError, TypeError):
			pass
	return 5


_client = None


def client():
	global _client
	if _client is None:
		here = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
		_client = Client(os.path.join(here, ".ca", "root.crt"))
	return _client