116 lines
3.6 KiB
Python
116 lines
3.6 KiB
Python
"""Low-level Docker socket client."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import socket
|
|
from pathlib import Path
|
|
from urllib.parse import urlparse
|
|
|
|
|
|
def _decode_chunked_bytes(data: bytes) -> bytes:
|
|
result = bytearray()
|
|
pos = 0
|
|
while pos < len(data):
|
|
end = data.find(b"\r\n", pos)
|
|
if end < 0:
|
|
break
|
|
try:
|
|
size = int(data[pos:end].split(b";")[0].strip(), 16)
|
|
except ValueError:
|
|
break
|
|
if size == 0:
|
|
break
|
|
pos = end + 2
|
|
result.extend(data[pos:pos + size])
|
|
pos += size + 2
|
|
return bytes(result)
|
|
|
|
|
|
def _docker_available() -> bool:
|
|
endpoint = os.environ.get("DOCKER_HOST", "")
|
|
if endpoint.startswith("tcp://"):
|
|
return bool(urlparse(endpoint).hostname and urlparse(endpoint).port)
|
|
return Path(os.environ.get("DOCKER_SOCKET", "/var/run/docker.sock")).exists()
|
|
|
|
|
|
def _docker_connection() -> socket.socket:
|
|
endpoint = os.environ.get("DOCKER_HOST", "")
|
|
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) if endpoint.startswith("tcp://") else socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
if endpoint.startswith("tcp://"):
|
|
parsed = urlparse(endpoint)
|
|
if not parsed.hostname or not parsed.port:
|
|
sock.close()
|
|
raise RuntimeError("Invalid DOCKER_HOST; expected tcp://host:port")
|
|
sock.connect((parsed.hostname, parsed.port))
|
|
else:
|
|
sock.connect(os.environ.get("DOCKER_SOCKET", "/var/run/docker.sock"))
|
|
return sock
|
|
|
|
|
|
def _docker_get_json(path: str) -> tuple[int, dict | list | None]:
|
|
if not _docker_available():
|
|
raise RuntimeError("Docker control endpoint is unavailable")
|
|
request = (
|
|
f"GET {path} HTTP/1.1\r\n"
|
|
"Host: docker\r\n"
|
|
"Connection: close\r\n\r\n"
|
|
).encode("utf-8")
|
|
chunks: list[bytes] = []
|
|
with _docker_connection() as sock:
|
|
sock.settimeout(10)
|
|
sock.connect(sock_path)
|
|
sock.sendall(request)
|
|
while True:
|
|
data = sock.recv(65536)
|
|
if not data:
|
|
break
|
|
chunks.append(data)
|
|
raw = b"".join(chunks)
|
|
header_end = raw.find(b"\r\n\r\n")
|
|
if header_end < 0:
|
|
raise RuntimeError("Invalid Docker HTTP response")
|
|
header_str = raw[:header_end].decode("utf-8", errors="replace")
|
|
body_bytes = raw[header_end + 4:]
|
|
status_line = header_str.splitlines()[0]
|
|
m = re.match(r"HTTP/\S+\s+(\d+)", status_line)
|
|
if not m:
|
|
raise RuntimeError(f"Invalid Docker status line: {status_line!r}")
|
|
code = int(m.group(1))
|
|
if "transfer-encoding: chunked" in header_str.lower():
|
|
body_bytes = _decode_chunked_bytes(body_bytes)
|
|
if not body_bytes.strip():
|
|
return code, None
|
|
return code, json.loads(body_bytes)
|
|
|
|
|
|
def _docker_post(path: str) -> tuple[int, str]:
|
|
if not _docker_available():
|
|
raise RuntimeError("Docker control endpoint is unavailable")
|
|
|
|
request = (
|
|
f"POST {path} HTTP/1.1\r\n"
|
|
"Host: docker\r\n"
|
|
"Content-Length: 0\r\n"
|
|
"Connection: close\r\n\r\n"
|
|
).encode("utf-8")
|
|
|
|
chunks = []
|
|
with _docker_connection() as sock:
|
|
sock.settimeout(20)
|
|
sock.connect(sock_path)
|
|
sock.sendall(request)
|
|
while True:
|
|
data = sock.recv(8192)
|
|
if not data:
|
|
break
|
|
chunks.append(data)
|
|
|
|
raw = b"".join(chunks).decode("utf-8", errors="replace")
|
|
status_line = raw.splitlines()[0] if raw else ""
|
|
match = re.match(r"HTTP/\S+\s+(\d+)", status_line)
|
|
if not match:
|
|
raise RuntimeError("Invalid Docker API response")
|
|
return int(match.group(1)), raw
|