1import http.client
2import io
3import json
4import os
5import socket
6import ssl
7import time
8import urllib.error
9import urllib.parse
10
11
12class LoopbackHTTPS(http.client.HTTPSConnection):
13 def connect(self):
14 sock = socket.create_connection(("127.0.0.1", self.port), self.timeout)
15 self.sock = self._context.wrap_socket(sock, server_hostname=self.host)
16
17
18class Keycloak:
19 def configure_profile(self):
20 profile = self.request("/admin/realms/master/users/profile")
21 if not any(attribute["name"] == "picture" for attribute in profile["attributes"]):
22 profile["attributes"].append({"name": "picture", "displayName": "Profile picture",
23 "permissions": {"view": ["admin", "user"], "edit": ["admin", "user"]},
24 "validations": {"length": {"max": 8192}}})
25 self.request("/admin/realms/master/users/profile", "PUT", profile)
26
27 def __init__(self, host, password, attempts=300, cafile=None):
28 self.host = host
29 self.context = ssl.create_default_context(
30 cafile=cafile or ("/var/lib/caddy/.local/share/caddy/pki/authorities/local/root.crt"
31 if host.endswith(".test") else None)
32 )
33 self.token = None
34 for attempt in range(attempts):
35 try:
36 body = urllib.parse.urlencode({
37 "client_id": "admin-cli",
38 "grant_type": "password",
39 "username": "admin",
40 "password": password,
41 }).encode()
42 self.token = self.request("/realms/master/protocol/openid-connect/token", "POST", body)["access_token"]
43 break
44 except urllib.error.HTTPError as error:
45 if error.code not in (502, 503, 504) or attempt == attempts - 1:
46 raise
47 time.sleep(2)
48 except (ConnectionError, TimeoutError, http.client.HTTPException, ssl.SSLError):
49 if attempt == attempts - 1:
50 raise
51 time.sleep(2)
52
53 def request(self, path, method="GET", body=None, full=False):
54 headers = {}
55 if self.token:
56 headers["Authorization"] = "Bearer " + self.token
57 if isinstance(body, (dict, list)):
58 body = json.dumps(body).encode()
59 headers["Content-Type"] = "application/json"
60 elif body is not None:
61 headers["Content-Type"] = "application/x-www-form-urlencoded"
62 connection = LoopbackHTTPS(
63 self.host,
64 timeout=int(os.getenv("STUDIO_API_TIMEOUT", "10")),
65 context=self.context,
66 )
67 try:
68 connection.request(method, path, body=body, headers=headers)
69 response = connection.getresponse()
70 content = response.read(16 * 1024 * 1024 + 1)
71 if len(content) > 16 * 1024 * 1024:
72 raise ValueError("Keycloak response exceeds 16 MiB")
73 if response.status >= 400:
74 raise urllib.error.HTTPError(
75 "https://" + self.host + path, response.status, response.reason,
76 response.headers, io.BytesIO(content),
77 )
78 value = json.loads(content) if content else None
79 if full:
80 return {"body": value, "id": response.getheader("Location", "").rsplit("/", 1)[-1] or None}
81 return value
82 finally:
83 connection.close()