149 lines
5.1 KiB
Python
149 lines
5.1 KiB
Python
"""Kubernetes resources that bind an OpenSandbox Kata guest to SPIRE."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
import ssl
|
|
from urllib.parse import quote
|
|
|
|
from aiohttp import ClientResponseError, ClientSession, TCPConnector
|
|
|
|
|
|
class SandboxKubernetesClient:
|
|
def __init__(
|
|
self,
|
|
*,
|
|
api_url: str,
|
|
token_file: Path,
|
|
ca_file: Path,
|
|
namespace: str,
|
|
) -> None:
|
|
self.api_url = api_url.rstrip("/")
|
|
self.token_file = token_file
|
|
self.ca_file = ca_file
|
|
self.namespace = namespace
|
|
self.session: ClientSession | None = None
|
|
|
|
async def __aenter__(self) -> SandboxKubernetesClient:
|
|
context = ssl.create_default_context(cafile=self.ca_file)
|
|
self.session = ClientSession(
|
|
connector=TCPConnector(ssl=context),
|
|
headers={
|
|
"Authorization": f"Bearer {self.token_file.read_text().strip()}"
|
|
},
|
|
raise_for_status=True,
|
|
)
|
|
return self
|
|
|
|
async def __aexit__(self, *_: object) -> None:
|
|
if self.session is not None:
|
|
await self.session.close()
|
|
|
|
async def get_batchsandbox(self, sandbox_id: str) -> dict[str, object] | None:
|
|
try:
|
|
response = await self._request(
|
|
"GET", f"{self._batchsandboxes_path()}/{quote(sandbox_id, safe='')}"
|
|
)
|
|
except ClientResponseError as error:
|
|
if error.status == 404:
|
|
return None
|
|
raise
|
|
return await response.json()
|
|
|
|
async def list_batchsandboxes(self) -> list[dict[str, object]]:
|
|
response = await self._request("GET", self._batchsandboxes_path())
|
|
document = await response.json()
|
|
items = document.get("items") if isinstance(document, dict) else None
|
|
return [item for item in items if isinstance(item, dict)] if isinstance(items, list) else []
|
|
|
|
async def get_pod(self, name: str) -> dict[str, object] | None:
|
|
try:
|
|
response = await self._request(
|
|
"GET", f"{self._pods_path()}/{quote(name, safe='')}"
|
|
)
|
|
except ClientResponseError as error:
|
|
if error.status == 404:
|
|
return None
|
|
raise
|
|
return await response.json()
|
|
|
|
async def create_entry(self, manifest: dict[str, object]) -> dict[str, object]:
|
|
response = await self._request(
|
|
"POST", self._entries_path(), json=manifest
|
|
)
|
|
return await response.json()
|
|
|
|
async def list_entries(self) -> list[dict[str, object]]:
|
|
response = await self._request(
|
|
"GET",
|
|
f"{self._entries_path()}?labelSelector="
|
|
"app.kubernetes.io%2Fcomponent%3Dopensandbox-identity",
|
|
)
|
|
document = await response.json()
|
|
items = document.get("items") if isinstance(document, dict) else None
|
|
return [item for item in items if isinstance(item, dict)] if isinstance(items, list) else []
|
|
|
|
async def get_entry(self, name: str) -> dict[str, object] | None:
|
|
try:
|
|
response = await self._request(
|
|
"GET", f"{self._entries_path()}/{quote(name, safe='')}"
|
|
)
|
|
except ClientResponseError as error:
|
|
if error.status == 404:
|
|
return None
|
|
raise
|
|
return await response.json()
|
|
|
|
async def delete_entry(self, name: str) -> None:
|
|
try:
|
|
response = await self._request(
|
|
"DELETE",
|
|
f"{self._entries_path()}/{quote(name, safe='')}",
|
|
json={"propagationPolicy": "Background"},
|
|
)
|
|
response.release()
|
|
except ClientResponseError as error:
|
|
if error.status != 404:
|
|
raise
|
|
|
|
async def _request(self, method: str, path: str, **kwargs: object):
|
|
if self.session is None:
|
|
raise RuntimeError("SandboxKubernetesClient is not open")
|
|
return await self.session.request(method, f"{self.api_url}{path}", **kwargs)
|
|
|
|
def _batchsandboxes_path(self) -> str:
|
|
return (
|
|
"/apis/sandbox.opensandbox.io/v1alpha1/namespaces/"
|
|
f"{quote(self.namespace, safe='')}/batchsandboxes"
|
|
)
|
|
|
|
def _pods_path(self) -> str:
|
|
return f"/api/v1/namespaces/{quote(self.namespace, safe='')}/pods"
|
|
|
|
@staticmethod
|
|
def _entries_path() -> str:
|
|
return "/apis/spire.spiffe.io/v1alpha1/clusterstaticentries"
|
|
|
|
|
|
def allocated_pod_name(batchsandbox: dict[str, object]) -> str | None:
|
|
metadata = batchsandbox.get("metadata")
|
|
if not isinstance(metadata, dict):
|
|
return None
|
|
annotations = metadata.get("annotations")
|
|
if not isinstance(annotations, dict):
|
|
return None
|
|
raw = annotations.get("sandbox.opensandbox.io/alloc-status")
|
|
if not isinstance(raw, str):
|
|
return None
|
|
try:
|
|
allocation = json.loads(raw)
|
|
except json.JSONDecodeError:
|
|
return None
|
|
if not isinstance(allocation, dict):
|
|
return None
|
|
pods = allocation.get("pods")
|
|
if not isinstance(pods, list) or len(pods) != 1 or not isinstance(pods[0], str):
|
|
return None
|
|
return pods[0]
|