Skip to content

Instantly share code, notes, and snippets.

@WuSiYu

WuSiYu/oci_stream_mirror.py

Last active July 13, 2026 14:48
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save WuSiYu/9b577da02052072c8526ceb3da921ec3 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save WuSiYu/9b577da02052072c8526ceb3da921ec3 to your computer and use it in GitHub Desktop.
Download ZIP
OCI Streaming Mirror: Docker/OCI registry cache-less reverse proxy
Raw
oci_stream_mirror.py
#!/usr/bin/env python3
"""
OCI Streaming Mirror POC
A single-file proof-of-concept for running a stateless Docker/OCI registry
streaming reverse proxy. It lets Docker pull images through custom mirror
domains such as:
docker pull hub.mirrors.example.com/library/ubuntu:24.04
docker pull ghcr.mirrors.example.com/modeltc/lightllm:main
docker pull nvcr.mirrors.example.com/nvidia/cuda:12.4.1-base-ubuntu22.04
The proxy does not cache image layers on disk. It forwards OCI Registry API
requests to the selected upstream registry, follows upstream blob redirects
server-side, and streams the response back to the Docker client.
Default upstream host prefixes:
hub.<base-domain> -> https://registry-1.docker.io
ghcr.<base-domain> -> https://ghcr.io
nvcr.<base-domain> -> https://nvcr.io
quay.<base-domain> -> https://quay.io
k8s.<base-domain> -> https://registry.k8s.io
gcr.<base-domain> -> https://gcr.io
mcr.<base-domain> -> https://mcr.microsoft.com
Basic usage with automatic HTTPS certificates:
sudo python3 oci_stream_mirror.py \
--base-domain mirrors.example.com \
--email admin@example.com \
--renew-loop
If --email is omitted, certbot is invoked in no-email mode:
sudo python3 oci_stream_mirror.py \
--base-domain mirrors.example.com
Run behind an existing HTTPS reverse proxy, such as nginx or Caddy:
sudo python3 oci_stream_mirror.py \
--base-domain mirrors.example.com \
--http-only \
--http-port 8080 \
--skip-certbot \
--trust-forwarded-for
Optional path-token authentication:
sudo python3 oci_stream_mirror.py \
--base-domain mirrors.example.com \
--path-auth mysecret
Then pull images with the token inserted after the host:
docker pull ghcr.mirrors.example.com/mysecret/modeltc/lightllm:main
docker pull hub.mirrors.example.com/mysecret/library/ubuntu:24.04
Serve an informational page on the base domain itself:
sudo python3 oci_stream_mirror.py \
--base-domain mirrors.example.com \
--serve-base-domain
This requires DNS records for both the base domain and the wildcard:
mirrors.example.com A/AAAA <server IP>
*.mirrors.example.com A/AAAA <server IP>
Serve a static directory on the base domain instead of the built-in page:
sudo python3 oci_stream_mirror.py \
--base-domain mirrors.example.com \
--web-root /var/www/oci-mirror
Install as a systemd service:
sudo python3 oci_stream_mirror.py \
--install \
--base-domain mirrors.example.com \
--path-auth mysecret \
--renew-loop
The installer copies the current script to:
/opt/oci_stream_mirror.py
and writes a systemd unit to:
/etc/systemd/system/oci-stream-mirror.service
The generated service embeds the current command-line arguments, except
installer-only options such as --install, --uninstall, and --service-name.
The installer does not start the service automatically. After installation:
sudo systemctl enable --now oci-stream-mirror
sudo journalctl -u oci-stream-mirror -f
sudo systemctl restart oci-stream-mirror
Uninstall the systemd service and remove /opt/oci_stream_mirror.py:
sudo python3 oci_stream_mirror.py --uninstall
Using non-standard service ports:
sudo python3 oci_stream_mirror.py \
--base-domain mirrors.example.com \
--http-port 8080 \
--https-port 8443
If port 80 is already used by another web server, certbot HTTP-01 validation
can use that server's webroot:
sudo python3 oci_stream_mirror.py \
--base-domain mirrors.example.com \
--no-http-server \
--https-port 8443 \
--acme-webroot /var/www/acme
The existing web server must serve:
/.well-known/acme-challenge/
from the same webroot.
Notes:
- Docker Hub official images need the explicit "library/" namespace when using
a custom registry host.
- This proxy only supports pull-style requests. Push/write methods are rejected.
- Mirror hosts only handle /v2/, /v2/*, /__oci_auth/*, and ACME challenge paths.
- With --serve-base-domain, the base domain serves an informational page.
- With --web-root, the base domain serves static files from that directory.
- This does not bypass upstream registry rate limits, because nothing is cached.
- Do not expose this service publicly without IP allowlisting, VPN access, or
path-token authentication.
"""
from __future__ import annotations
import argparse
import base64
import hashlib
import hmac
import html
import http.client
import errno
import ipaddress
import mimetypes
import os
import posixpath
import re
import secrets
import shlex
import shutil
import signal
import socket
import socketserver
import ssl
import subprocess
import sys
import textwrap
import threading
import time
from dataclasses import dataclass, field
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from typing import Dict, Iterable, List, Optional, Tuple
from urllib.parse import parse_qsl, unquote, urlencode, urljoin, urlsplit, urlunsplit
DEFAULT_UPSTREAMS = {
# Docker Hub official images need hub.<base>/library/ubuntu:24.04
"hub": "https://registry-1.docker.io",
"ghcr": "https://ghcr.io",
"nvcr": "https://nvcr.io",
"quay": "https://quay.io",
"k8s": "https://registry.k8s.io",
"gcr": "https://gcr.io",
"mcr": "https://mcr.microsoft.com",
}
HOP_BY_HOP_HEADERS = {
"connection",
"keep-alive",
"proxy-authenticate",
"proxy-authorization",
"te",
"trailer",
"transfer-encoding",
"upgrade",
}
REALM_RE = re.compile(r'realm="([^"]+)"', re.IGNORECASE)
SAFE_TOKEN_RE = re.compile(r"^[A-Za-z0-9._~-]{1,64}$")
DNS_LABEL_RE = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$")
HOST_PORT_RE = re.compile(r"^([A-Za-z0-9.-]+)(?::([0-9]{1,5}))?$")
CERT_NAME_RE = re.compile(r"^[A-Za-z0-9_.-]{1,128}$")
AUTH_SIG_HEX_LEN = 32
MAX_REDIRECTS = 8
COPY_CHUNK_SIZE = 1024 * 1024
DEFAULT_PULL_LOG_IDLE_SECONDS = 5.0
# Some registries, notably GHCR, may return "manifest unknown: OCI index found,
# but Accept header does not support OCI indexes" if /manifests/<ref> does not
# explicitly advertise OCI index support. This proxy defensively unions the
# client's Accept value with common Docker/OCI manifest media types for manifest
# requests only. Blob requests are untouched.
MANIFEST_ACCEPT_TYPES = [
"application/vnd.docker.distribution.manifest.v2+json",
"application/vnd.docker.distribution.manifest.list.v2+json",
"application/vnd.oci.image.manifest.v1+json",
"application/vnd.oci.image.index.v1+json",
"application/vnd.oci.artifact.manifest.v1+json",
"application/vnd.docker.distribution.manifest.v1+json",
"application/vnd.docker.distribution.manifest.v1+prettyjws",
]
@dataclass
class Upstream:
alias: str
public_host: str
registry_base: str
@dataclass
class AppConfig:
base_domain: str
bind: str
http_port: int
https_port: int
acme_webroot: str
cert_name: str
cert_file: str
key_file: str
serve_base_domain: bool
web_root: Optional[str]
upstream_by_host: Dict[str, Upstream]
allowed_cidrs: List[ipaddress._BaseNetwork]
upstream_timeout: int
client_timeout: int
tls_handshake_timeout: int
max_clients: int
pull_log_idle: float
verbose: bool
http_only: bool
external_scheme: str
trust_forwarded_for: bool
path_auth_enabled: bool
path_token: Optional[str]
auth_relay_secret: bytes
ban_threshold: int
ban_seconds: int
auth_failures: Dict[str, int] = field(default_factory=dict)
banned_until: Dict[str, float] = field(default_factory=dict)
auth_lock: threading.Lock = field(default_factory=threading.Lock)
CONFIG: Optional[AppConfig] = None
TLS_CONTEXT: Optional[ssl.SSLContext] = None
STOP_EVENT = threading.Event()
SERVERS: List[ThreadingHTTPServer] = []
SERVER_THREADS: List[threading.Thread] = []
def log(msg: str) -> None:
print(time.strftime("[%Y-%m-%d %H:%M:%S] ") + msg, flush=True)
def one_line_exception(exc: BaseException, max_len: int = 180) -> str:
msg = str(exc) or repr(exc)
msg = msg.replace("\n", " ").replace("\r", " ")
if len(msg) > max_len:
msg = msg[: max_len - 3] + "..."
return f"{exc.__class__.__name__}: {msg}"
def is_quiet_network_error(exc: BaseException) -> bool:
"""Return True for common client-side disconnect / TLS scanner noise.
These are not application errors and should not produce a traceback in
normal operation. In verbose mode they are logged as a single line.
"""
if isinstance(exc, (BrokenPipeError, ConnectionResetError, TimeoutError, socket.timeout, ssl.SSLError)):
return True
if isinstance(exc, OSError):
noisy_errno = {
errno.EPIPE,
errno.ECONNRESET,
errno.ECONNABORTED,
errno.ETIMEDOUT,
}
return getattr(exc, "errno", None) in noisy_errno
return False
def log_quiet_network_error(where: str, client: object, exc: BaseException) -> None:
cfg = CONFIG
if not (cfg and cfg.verbose):
return
if isinstance(client, tuple) and client:
client_s = str(client[0])
else:
client_s = str(client)
log(f"network-close where={where} client={safe_log_value(client_s, 128)} error={safe_log_value(one_line_exception(exc), 256)}")
def die(msg: str, code: int = 2) -> None:
print(f"fatal: {msg}", file=sys.stderr)
sys.exit(code)
def safe_log_value(value: str, max_len: int = 512) -> str:
value = value.replace("\n", "\\n").replace("\r", "\\r")
value = re.sub(r"\s+", " ", value).strip()
if len(value) > max_len:
value = value[: max_len - 1] + "…"
return value
def format_size(num_bytes: int) -> str:
value = float(max(0, num_bytes))
for unit in ("B", "KiB", "MiB", "GiB", "TiB", "PiB"):
if value < 1024.0 or unit == "PiB":
if unit == "B":
return f"{int(value)} {unit}"
return f"{value:.2f} {unit}"
value /= 1024.0
return f"{num_bytes} B"
def format_speed(num_bytes: int, seconds: float) -> str:
seconds = max(seconds, 0.001)
return format_size(int(num_bytes / seconds)) + "/s"
def parse_manifest_request_path(path: str) -> Optional[Tuple[str, str]]:
if not path.startswith("/v2/"):
return None
marker = "/manifests/"
idx = path.find(marker, len("/v2/"))
if idx < 0:
return None
repo = path[len("/v2/") : idx]
ref = path[idx + len(marker) :]
if not repo or not ref or "/" in ref:
return None
return unquote(repo), unquote(ref)
def parse_blob_request_path(path: str) -> Optional[Tuple[str, str]]:
if not path.startswith("/v2/"):
return None
marker = "/blobs/"
idx = path.find(marker, len("/v2/"))
if idx < 0:
return None
repo = path[len("/v2/") : idx]
digest = path[idx + len(marker) :]
if not repo or not digest or "/" in digest:
return None
return unquote(repo), unquote(digest)
def format_image_ref(public_host: str, repo: str, ref: str) -> str:
if ref.startswith("sha256:") or ref.startswith("sha512:"):
return f"{public_host}/{repo}@{ref}"
return f"{public_host}/{repo}:{ref}"
@dataclass
class StreamStats:
response_status: Optional[int] = None
body_bytes: int = 0
@dataclass
class PullSession:
key: Tuple[str, str, str]
client_ip: str
public_host: str
repo: str
image: str
started_at: float
last_activity: float
bytes_total: int = 0
active_blobs: int = 0
failed: bool = False
finalized: bool = False
timer: Optional[threading.Timer] = None
class PullLogManager:
"""
Best-effort Docker pull session logger.
Docker/OCI Registry does not expose a single "pull finished" request. A
client normally performs manifest requests followed by several concurrent
blob requests. This manager groups requests by client IP + mirror host +
repository, logs one START line at the first manifest request, accumulates
streamed blob bytes, and logs one END line after a short idle period once no
blob transfer is active.
"""
def __init__(self) -> None:
self.lock = threading.Lock()
self.sessions: Dict[Tuple[str, str, str], PullSession] = {}
def idle_seconds(self) -> float:
cfg = CONFIG
if cfg is None:
return DEFAULT_PULL_LOG_IDLE_SECONDS
return max(0.5, float(cfg.pull_log_idle))
def start_manifest(self, client_ip: str, public_host: str, repo: str, ref: str) -> Tuple[str, str, str]:
now = time.time()
key = (client_ip, public_host, repo)
image = format_image_ref(public_host, repo, ref)
with self.lock:
existing = self.sessions.get(key)
if existing and not existing.finalized:
# Keep the first user-visible image name for this pull. Docker
# often requests a tag manifest first and then a platform
# manifest by digest; logging both would create duplicate pull
# sessions for one docker pull. Concurrent pulls of different
# tags in the same repo from the same client may be merged,
# which is acceptable for this lightweight best-effort logger.
existing.last_activity = now
self._cancel_timer_locked(existing)
return key
session = PullSession(
key=key,
client_ip=client_ip,
public_host=public_host,
repo=repo,
image=image,
started_at=now,
last_activity=now,
)
self.sessions[key] = session
log(f"PULL START client={safe_log_value(client_ip)} image={safe_log_value(image)}")
return key
def start_blob(self, client_ip: str, public_host: str, repo: str, digest: str) -> Tuple[str, str, str]:
now = time.time()
key = (client_ip, public_host, repo)
with self.lock:
session = self.sessions.get(key)
if session is None or session.finalized:
short_digest = digest if len(digest) <= 32 else digest[:32] + "…"
image = f"{public_host}/{repo}@{short_digest}"
session = PullSession(
key=key,
client_ip=client_ip,
public_host=public_host,
repo=repo,
image=image,
started_at=now,
last_activity=now,
)
self.sessions[key] = session
log(f"PULL START client={safe_log_value(client_ip)} image={safe_log_value(image)}")
session.active_blobs += 1
session.last_activity = now
self._cancel_timer_locked(session)
return key
def finish_manifest(self, key: Tuple[str, str, str], status: Optional[int]) -> None:
now = time.time()
with self.lock:
session = self.sessions.get(key)
if not session or session.finalized:
return
session.last_activity = now
if status is not None and status >= 400:
session.failed = True
if session.active_blobs == 0:
self._schedule_finalize_locked(session)
def finish_blob(self, key: Tuple[str, str, str], stats: StreamStats, failed: bool = False) -> None:
now = time.time()
with self.lock:
session = self.sessions.get(key)
if not session or session.finalized:
return
session.bytes_total += max(0, stats.body_bytes)
session.last_activity = now
if failed or (stats.response_status is not None and stats.response_status >= 400):
session.failed = True
if session.active_blobs > 0:
session.active_blobs -= 1
if session.active_blobs == 0:
self._schedule_finalize_locked(session)
def fail_request(self, key: Tuple[str, str, str], stats: Optional[StreamStats] = None) -> None:
now = time.time()
with self.lock:
session = self.sessions.get(key)
if not session or session.finalized:
return
if stats is not None:
session.bytes_total += max(0, stats.body_bytes)
session.failed = True
session.last_activity = now
if session.active_blobs > 0:
session.active_blobs -= 1
if session.active_blobs == 0:
self._schedule_finalize_locked(session)
def finalize_if_idle(self, key: Tuple[str, str, str]) -> None:
now = time.time()
with self.lock:
session = self.sessions.get(key)
if not session or session.finalized:
return
if session.active_blobs > 0:
return
idle = now - session.last_activity
delay = self.idle_seconds()
if idle + 0.05 < delay and not STOP_EVENT.is_set():
self._schedule_finalize_locked(session)
return
self._finalize_locked(session, now)
def finalize_all(self) -> None:
now = time.time()
with self.lock:
for session in list(self.sessions.values()):
if not session.finalized:
self._finalize_locked(session, now)
def _cancel_timer_locked(self, session: PullSession) -> None:
if session.timer is not None:
try:
session.timer.cancel()
except Exception:
pass
session.timer = None
def _schedule_finalize_locked(self, session: PullSession) -> None:
self._cancel_timer_locked(session)
if STOP_EVENT.is_set():
self._finalize_locked(session, time.time())
return
timer = threading.Timer(self.idle_seconds(), self.finalize_if_idle, args=(session.key,))
timer.daemon = True
session.timer = timer
timer.start()
def _finalize_locked(self, session: PullSession, now: float) -> None:
session.finalized = True
self._cancel_timer_locked(session)
self.sessions.pop(session.key, None)
duration = max(0.001, now - session.started_at)
status = "failed" if session.failed else "ok"
log(
"PULL END "
f"client={safe_log_value(session.client_ip)} "
f"image={safe_log_value(session.image)} "
f"status={status} "
f"size={format_size(session.bytes_total)} "
f"duration={duration:.2f}s "
f"speed={format_speed(session.bytes_total, duration)}"
)
PULL_LOG = PullLogManager()
def validate_dns_name(name: str, what: str) -> str:
value = name.strip().lower().rstrip(".")
if len(value) > 253 or "." not in value:
die(f"{what} should be a valid DNS name, e.g. mirrors.example.com")
labels = value.split(".")
if any(not DNS_LABEL_RE.fullmatch(label) for label in labels):
die(f"{what} contains an invalid DNS label")
return value
def validate_upstream_alias(alias: str) -> str:
value = alias.strip().lower().rstrip(".")
if not DNS_LABEL_RE.fullmatch(value):
die(f"bad upstream alias {alias!r}; use a valid single DNS label, e.g. ghcr")
return value
def validate_cert_name(name: str) -> str:
value = name.strip()
if not CERT_NAME_RE.fullmatch(value):
die("--cert-name contains unsupported characters")
return value
def validate_port(port: int, what: str) -> int:
if port < 1 or port > 65535:
die(f"{what} must be in 1..65535")
return port
def parse_safe_host_header(host_header: str) -> Optional[Tuple[str, str]]:
# This POC is designed for normal DNS hostnames, not IPv6 literals.
m = HOST_PORT_RE.fullmatch(host_header.strip())
if not m:
return None
host = m.group(1).lower().rstrip(".")
labels = host.split(".")
if len(host) > 253 or any(not DNS_LABEL_RE.fullmatch(label) for label in labels):
return None
port = m.group(2)
if port is None:
return host, host
n = int(port)
if n < 1 or n > 65535:
return None
return host, f"{host}:{n}"
def b64url_encode(s: str) -> str:
return base64.urlsafe_b64encode(s.encode("utf-8")).decode("ascii").rstrip("=")
def b64url_decode(s: str) -> str:
s += "=" * (-len(s) % 4)
return base64.urlsafe_b64decode(s.encode("ascii")).decode("utf-8")
def sign_auth_realm(encoded_realm: str) -> str:
cfg = CONFIG
assert cfg is not None
return hmac.new(
cfg.auth_relay_secret,
encoded_realm.encode("ascii"),
hashlib.sha256,
).hexdigest()[:AUTH_SIG_HEX_LEN]
def make_auth_relay_token(upstream_realm: str) -> str:
encoded = b64url_encode(upstream_realm)
return f"{encoded}.{sign_auth_realm(encoded)}"
def verify_auth_relay_token(token: str) -> Optional[str]:
if "." not in token:
return None
encoded, sig = token.rsplit(".", 1)
if not re.fullmatch(r"[A-Za-z0-9_-]+", encoded):
return None
if not re.fullmatch(r"[a-f0-9]+", sig):
return None
expected = sign_auth_realm(encoded)
if not hmac.compare_digest(sig, expected):
return None
try:
realm = b64url_decode(encoded)
except Exception:
return None
p = urlsplit(realm)
# Registry token endpoints should be HTTPS URLs returned by known upstream registries.
# HMAC prevents external clients from choosing arbitrary URLs.
if p.scheme != "https" or not p.netloc:
return None
return realm
def merge_query(base_url: str, extra_query: str) -> str:
p = urlsplit(base_url)
q = []
if p.query:
q.extend(parse_qsl(p.query, keep_blank_values=True))
if extra_query:
q.extend(parse_qsl(extra_query, keep_blank_values=True))
return urlunsplit((p.scheme, p.netloc, p.path or "/", urlencode(q, doseq=True), ""))
def rewrite_www_authenticate(value: str, public_netloc: str) -> str:
cfg = CONFIG
assert cfg is not None
if "bearer" not in value.lower():
return value
def repl(match: re.Match[str]) -> str:
upstream_realm = match.group(1)
relay_token = make_auth_relay_token(upstream_realm)
return f'realm="{cfg.external_scheme}://{public_netloc}/__oci_auth/{relay_token}"'
return REALM_RE.sub(repl, value)
def get_client_ip(handler: BaseHTTPRequestHandler) -> str:
cfg = CONFIG
assert cfg is not None
if cfg.trust_forwarded_for:
xff = handler.headers.get("X-Forwarded-For", "")
if xff:
first = xff.split(",", 1)[0].strip()
if first:
return first
xri = handler.headers.get("X-Real-IP", "").strip()
if xri:
return xri
return handler.client_address[0]
def is_client_allowed(client_ip: str) -> bool:
cfg = CONFIG
assert cfg is not None
if not cfg.allowed_cidrs:
return True
try:
ip = ipaddress.ip_address(client_ip)
except ValueError:
return False
return any(ip in net for net in cfg.allowed_cidrs)
def is_ip_banned(client_ip: str) -> bool:
cfg = CONFIG
assert cfg is not None
if not cfg.path_auth_enabled:
return False
now = time.time()
with cfg.auth_lock:
until = cfg.banned_until.get(client_ip)
if until is None:
return False
if now >= until:
cfg.banned_until.pop(client_ip, None)
cfg.auth_failures.pop(client_ip, None)
return False
return True
def record_auth_failure(client_ip: str) -> None:
cfg = CONFIG
assert cfg is not None
if not cfg.path_auth_enabled:
return
with cfg.auth_lock:
n = cfg.auth_failures.get(client_ip, 0) + 1
if n >= cfg.ban_threshold:
cfg.banned_until[client_ip] = time.time() + cfg.ban_seconds
cfg.auth_failures[client_ip] = 0
log(f"temporarily banned {client_ip} for {cfg.ban_seconds}s after {cfg.ban_threshold} bad path-token attempts")
else:
cfg.auth_failures[client_ip] = n
def clear_auth_failure(client_ip: str) -> None:
cfg = CONFIG
assert cfg is not None
if not cfg.path_auth_enabled:
return
with cfg.auth_lock:
cfg.auth_failures.pop(client_ip, None)
def copy_request_headers(src_headers, target_netloc: str, drop_authorization: bool = False) -> Dict[str, str]:
connection_tokens = set()
for v in src_headers.get_all("Connection", []):
connection_tokens.update(x.strip().lower() for x in v.split(",") if x.strip())
headers: Dict[str, str] = {}
for k, v in src_headers.items():
lk = k.lower()
if lk in HOP_BY_HOP_HEADERS or lk in connection_tokens or lk == "host":
continue
if drop_authorization and lk == "authorization":
continue
headers[k] = v
headers["Host"] = target_netloc
return headers
def get_header_ci(headers: Dict[str, str], name: str) -> Tuple[Optional[str], Optional[str]]:
wanted = name.lower()
for k, v in headers.items():
if k.lower() == wanted:
return k, v
return None, None
def set_header_ci(headers: Dict[str, str], name: str, value: str) -> None:
existing_key, _ = get_header_ci(headers, name)
headers[existing_key or name] = value
def ensure_manifest_accept(headers: Dict[str, str], upstream_path: str) -> None:
if "/manifests/" not in upstream_path:
return
_, existing = get_header_ci(headers, "Accept")
existing = existing or ""
values = [x.strip() for x in existing.split(",") if x.strip()]
seen = {x.split(";", 1)[0].strip().lower() for x in values}
for media_type in MANIFEST_ACCEPT_TYPES:
if media_type.lower() not in seen:
values.append(media_type)
seen.add(media_type.lower())
set_header_ci(headers, "Accept", ", ".join(values))
def add_forwarded_for(headers: Dict[str, str], client_ip: str) -> None:
prior = headers.get("X-Forwarded-For")
headers["X-Forwarded-For"] = f"{prior}, {client_ip}" if prior else client_ip
def is_allowed_oci_path(path: str) -> bool:
return path == "/v2" or path.startswith("/v2/") or path.startswith("/__oci_auth/")
def is_allowed_pull_method(method: str, path: str) -> bool:
if path.startswith("/__oci_auth/"):
return method in {"GET", "HEAD", "OPTIONS"}
return method in {"GET", "HEAD", "OPTIONS"}
def strip_path_token_for_upstream(path: str, client_ip: str) -> Optional[str]:
cfg = CONFIG
assert cfg is not None
if not cfg.path_auth_enabled:
return path
if path in {"/v2", "/v2/"}:
return path
if not path.startswith("/v2/"):
return path
rest = path[len("/v2/") :]
token, sep, repo_rest = rest.partition("/")
if token != cfg.path_token:
record_auth_failure(client_ip)
return None
clear_auth_failure(client_ip)
if not sep or not repo_rest:
return "/v2/"
return "/v2/" + repo_rest
def strip_path_token_from_scope_query(query: str) -> str:
cfg = CONFIG
assert cfg is not None
if not cfg.path_auth_enabled or not cfg.path_token or not query:
return query
changed = False
out = []
prefix = f"repository:{cfg.path_token}/"
for k, v in parse_qsl(query, keep_blank_values=True):
if k == "scope" and v.startswith(prefix):
v = "repository:" + v[len(prefix) :]
changed = True
out.append((k, v))
return urlencode(out, doseq=True) if changed else query
def mirror_example_path() -> str:
cfg = CONFIG
assert cfg is not None
return "<TOKEN>/" if cfg.path_auth_enabled else ""
def build_default_base_page(client_ip: str) -> bytes:
cfg = CONFIG
assert cfg is not None
base = html.escape(cfg.base_domain)
prefix = html.escape(mirror_example_path())
shown_client_ip = html.escape(client_ip)
examples = {
"hub": f"docker pull hub.{base}/{prefix}library/ubuntu:24.04",
"ghcr": f"docker pull ghcr.{base}/{prefix}modeltc/lightllm:main",
"nvcr": f"docker pull nvcr.{base}/{prefix}nvidia/cuda:12.4.1-base-ubuntu22.04",
"quay": f"docker pull quay.{base}/{prefix}prometheus/prometheus:latest",
"k8s": f"docker pull k8s.{base}/{prefix}pause:3.10",
"gcr": f"docker pull gcr.{base}/{prefix}distroless/static-debian12:latest",
"mcr": f"docker pull mcr.{base}/{prefix}hello-world",
}
rows = []
for host, up in sorted(cfg.upstream_by_host.items()):
alias = html.escape(up.alias)
mirror_host = html.escape(host)
target = html.escape(up.registry_base)
cmd = html.escape(examples.get(up.alias, f"docker pull {host}/{prefix}namespace/image:tag"))
rows.append(
f"<tr><td><code>{alias}</code></td>"
f"<td><code>{mirror_host}</code></td>"
f"<td><code>{target}</code></td>"
f"<td><pre>{cmd}</pre></td></tr>"
)
auth_note = ""
if cfg.path_auth_enabled:
auth_note = (
"<p><strong>Path authentication is enabled.</strong> "
"Replace <code>&lt;TOKEN&gt;</code> with the configured token. "
"The token is intentionally not shown on this page.</p>"
)
doc = f"""<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>OCI Streaming Mirror</title>
<style>
body{{font-family:-apple-system,BlinkMacSystemFont,Segoe UI,Roboto,sans-serif;max-width:1100px;margin:40px auto;padding:0 20px;line-height:1.5}}
code,pre{{background:#f5f5f5;border-radius:6px}}
code{{padding:2px 5px}}
pre{{padding:8px;overflow:auto;margin:0}}
table{{border-collapse:collapse;width:100%;margin-top:16px}}
th,td{{border:1px solid #ddd;padding:8px;text-align:left;vertical-align:top}}
th{{background:#fafafa}}
.warn{{color:#9a6700}}
</style>
</head>
<body>
<h1>OCI Streaming Mirror</h1>
<p>This service is a stateless Docker/OCI registry streaming reverse proxy. It does not cache image layers.</p>
{auth_note}
<p>Docker Hub official images need the explicit <code>library/</code> namespace when using a custom registry host.</p>
<table>
<thead><tr><th>Prefix</th><th>Mirror host</th><th>Upstream</th><th>Example</th></tr></thead>
<tbody>
{''.join(rows)}
</tbody>
</table>
<p class="warn">Client IP: <code>{shown_client_ip}</code></p>
</body>
</html>
"""
return doc.encode("utf-8")
def safe_static_file_for_request(root: str, request_path: str) -> Optional[Path]:
root_path = Path(root).resolve()
raw_path = unquote(urlsplit(request_path).path)
if "\x00" in raw_path:
return None
rel = posixpath.normpath(raw_path.lstrip("/"))
if rel in (".", ""):
rel = "index.html"
candidate = (root_path / rel).resolve()
try:
candidate.relative_to(root_path)
except ValueError:
return None
if candidate.is_dir():
candidate = (candidate / "index.html").resolve()
try:
candidate.relative_to(root_path)
except ValueError:
return None
if not candidate.is_file():
return None
return candidate
class TimeoutAndTLSRequestMixin:
"""
Per-client timeouts and per-connection TLS handshake handling.
Important: do not wrap the listening socket with SSLContext.wrap_socket().
If TLS handshakes happen in the accept/serve_forever thread, a scanner can
open a TLS connection and never finish the handshake, blocking all new
accepts and making systemd stop wait forever in server.shutdown().
"""
def setup(self) -> None:
cfg = CONFIG
raw_sock = self.request
client_timeout = cfg.client_timeout if cfg is not None else 300
handshake_timeout = cfg.tls_handshake_timeout if cfg is not None else 10
ctx = getattr(self.server, "ssl_context", None)
if ctx is not None:
raw_sock.settimeout(handshake_timeout)
conn = ctx.wrap_socket(raw_sock, server_side=True, do_handshake_on_connect=False)
try:
conn.do_handshake()
except Exception:
conn.close()
raise
conn.settimeout(client_timeout)
else:
raw_sock.settimeout(client_timeout)
conn = raw_sock
self.connection = conn
self.rfile = self.connection.makefile("rb", self.rbufsize)
if self.wbufsize == 0:
self.wfile = socketserver._SocketWriter(self.connection)
else:
self.wfile = self.connection.makefile("wb", self.wbufsize)
class QuietBrokenPipeMixin:
def handle(self) -> None:
try:
super().handle()
except (BrokenPipeError, ConnectionResetError, TimeoutError, socket.timeout, ssl.SSLError, OSError) as exc:
log_quiet_network_error("handler", getattr(self, "client_address", "unknown"), exc)
return
class ReuseThreadingHTTPServer(ThreadingHTTPServer):
allow_reuse_address = True
daemon_threads = True
block_on_close = False
request_queue_size = 128
def __init__(self, server_address, RequestHandlerClass, *, ssl_context=None):
self.ssl_context = ssl_context
cfg = CONFIG
max_clients = cfg.max_clients if cfg is not None else 256
self._client_slots = threading.BoundedSemaphore(max_clients)
super().__init__(server_address, RequestHandlerClass)
def handle_error(self, request, client_address):
exc = sys.exc_info()[1]
if exc is not None and is_quiet_network_error(exc):
log_quiet_network_error("server", client_address, exc)
return
super().handle_error(request, client_address)
def process_request(self, request, client_address):
if not self._client_slots.acquire(blocking=False):
try:
request.close()
except Exception:
pass
return
try:
super().process_request(request, client_address)
except Exception:
self._client_slots.release()
raise
def process_request_thread(self, request, client_address):
try:
super().process_request_thread(request, client_address)
finally:
self._client_slots.release()
class ChallengeAndRedirectHandler(QuietBrokenPipeMixin, TimeoutAndTLSRequestMixin, BaseHTTPRequestHandler):
server_version = "oci-stream-mirror/0.8"
def log_message(self, fmt: str, *args) -> None:
cfg = CONFIG
if cfg and cfg.verbose:
super().log_message(fmt, *args)
def do_HEAD(self) -> None:
self._handle(send_body=False)
def do_GET(self) -> None:
self._handle(send_body=True)
def _handle(self, send_body: bool) -> None:
cfg = CONFIG
assert cfg is not None
client_ip = get_client_ip(self)
path = urlsplit(self.path).path
# ACME challenge must remain reachable even if allow-cidr/path-auth would reject normal pull traffic.
if path.startswith("/.well-known/acme-challenge/"):
rel = path.lstrip("/")
root = os.path.abspath(cfg.acme_webroot)
fp = os.path.abspath(os.path.join(root, rel))
if not fp.startswith(root + os.sep) or not os.path.isfile(fp):
self.send_error(404, "challenge not found")
return
self.send_response(200)
self.send_header("Content-Type", "text/plain")
self.send_header("Content-Length", str(os.path.getsize(fp)))
self.end_headers()
if send_body:
with open(fp, "rb") as f:
shutil.copyfileobj(f, self.wfile, COPY_CHUNK_SIZE)
return
if not is_client_allowed(client_ip):
self.send_error(403, "forbidden")
return
host_header = self.headers.get("Host", cfg.base_domain)
parsed_host = parse_safe_host_header(host_header)
if parsed_host is None:
self.send_error(400, "bad host header")
return
_, public_netloc = parsed_host
location = f"https://{public_netloc}{self.path}"
self.send_response(301)
self.send_header("Location", location)
self.send_header("Content-Length", "0")
self.end_headers()
class OCIProxyHandler(QuietBrokenPipeMixin, TimeoutAndTLSRequestMixin, BaseHTTPRequestHandler):
server_version = "oci-stream-mirror/0.8"
protocol_version = "HTTP/1.1"
def log_message(self, fmt: str, *args) -> None:
cfg = CONFIG
if cfg and cfg.verbose:
super().log_message(fmt, *args)
def do_GET(self) -> None:
self._handle_proxy()
def do_HEAD(self) -> None:
self._handle_proxy()
def do_OPTIONS(self) -> None:
self._handle_proxy()
def do_POST(self) -> None:
self._send_text(405, "method not allowed: this POC is pull-only\n")
def do_PUT(self) -> None:
self._send_text(405, "method not allowed: this POC is pull-only\n")
def do_PATCH(self) -> None:
self._send_text(405, "method not allowed: this POC is pull-only\n")
def do_DELETE(self) -> None:
self._send_text(405, "method not allowed: this POC is pull-only\n")
def _handle_proxy(self) -> None:
cfg = CONFIG
assert cfg is not None
client_ip = get_client_ip(self)
if not is_client_allowed(client_ip):
self.send_error(403, "forbidden")
return
if is_ip_banned(client_ip):
self.send_error(403, "temporarily banned")
return
host_header = self.headers.get("Host", "")
parsed_host = parse_safe_host_header(host_header)
if parsed_host is None:
self._send_text(400, "bad host header\n")
return
public_host, public_netloc = parsed_host
parsed = urlsplit(self.path)
pull_log_key: Optional[Tuple[str, str, str]] = None
pull_log_is_blob = False
if public_host == cfg.base_domain:
if cfg.serve_base_domain or cfg.web_root:
self._handle_base_domain(send_body=(self.command != "HEAD"))
else:
self._send_text(404, "base domain service is disabled; use --serve-base-domain or --web-root\n")
return
if not is_allowed_oci_path(parsed.path):
self._send_text(404, "not found: this endpoint only proxies OCI Registry /v2/* pulls\n")
return
if not is_allowed_pull_method(self.command, parsed.path):
self._send_text(405, "method not allowed: this POC is pull-only\n")
return
try:
if parsed.path.startswith("/__oci_auth/"):
if public_host not in cfg.upstream_by_host:
self._send_text(404, "unknown mirror host\n")
return
relay_token = parsed.path[len("/__oci_auth/") :].split("/", 1)[0]
upstream_realm = verify_auth_relay_token(relay_token)
if upstream_realm is None:
self._send_text(404, "not found\n")
return
auth_query = strip_path_token_from_scope_query(parsed.query)
target_url = merge_query(upstream_realm, auth_query)
else:
upstream = cfg.upstream_by_host.get(public_host)
if not upstream:
known = ", ".join(sorted(cfg.upstream_by_host.keys()))
self._send_text(404, f"unknown mirror host: {public_host}\nknown hosts: {known}\n")
return
upstream_path = strip_path_token_for_upstream(parsed.path, client_ip)
if upstream_path is None:
self._send_text(404, "not found\n")
return
manifest_info = parse_manifest_request_path(upstream_path)
if manifest_info is not None:
repo, ref = manifest_info
pull_log_key = PULL_LOG.start_manifest(client_ip, public_host, repo, ref)
pull_log_is_blob = False
else:
blob_info = parse_blob_request_path(upstream_path)
if blob_info is not None:
repo, digest = blob_info
pull_log_key = PULL_LOG.start_blob(client_ip, public_host, repo, digest)
pull_log_is_blob = True
upstream_uri = urlunsplit(("", "", upstream_path, parsed.query, ""))
target_url = upstream.registry_base.rstrip("/") + upstream_uri
if cfg.verbose:
log(f"{client_ip} {self.command} {public_host}{self.path} -> {target_url}")
stats = StreamStats()
self._forward_streaming(target_url, public_netloc, client_ip, stats=stats)
if pull_log_key is not None:
if pull_log_is_blob:
PULL_LOG.finish_blob(pull_log_key, stats)
else:
PULL_LOG.finish_manifest(pull_log_key, stats.response_status)
except BrokenPipeError:
if pull_log_key is not None:
if pull_log_is_blob:
PULL_LOG.fail_request(pull_log_key, stats if 'stats' in locals() else None)
else:
PULL_LOG.finish_manifest(pull_log_key, 499)
return
except Exception as e:
if pull_log_key is not None:
if pull_log_is_blob:
PULL_LOG.fail_request(pull_log_key, stats if 'stats' in locals() else None)
else:
PULL_LOG.finish_manifest(pull_log_key, 502)
self._send_text(502, f"proxy error: {type(e).__name__}: {e}\n")
def _handle_base_domain(self, send_body: bool) -> None:
cfg = CONFIG
assert cfg is not None
if self.command not in {"GET", "HEAD", "OPTIONS"}:
self._send_text(405, "method not allowed\n")
return
if self.command == "OPTIONS":
self.send_response(204)
self.send_header("Content-Length", "0")
self.end_headers()
return
if cfg.web_root:
fp = safe_static_file_for_request(cfg.web_root, self.path)
if fp is None:
self._send_text(404, "not found\n")
return
ctype = mimetypes.guess_type(str(fp))[0] or "application/octet-stream"
try:
size = fp.stat().st_size
self.send_response(200)
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(size))
self.end_headers()
if send_body:
with fp.open("rb") as f:
shutil.copyfileobj(f, self.wfile, COPY_CHUNK_SIZE)
return
except OSError:
self._send_text(404, "not found\n")
return
data = build_default_base_page(get_client_ip(self))
self.send_response(200)
self.send_header("Content-Type", "text/html; charset=utf-8")
self.send_header("Content-Length", str(len(data)))
self.end_headers()
if send_body:
self.wfile.write(data)
def _send_text(self, status: int, text: str) -> None:
data = text.encode("utf-8")
self.send_response(status)
self.send_header("Content-Type", "text/plain; charset=utf-8")
self.send_header("Content-Length", str(len(data)))
self.end_headers()
if self.command != "HEAD":
self.wfile.write(data)
def _read_small_body(self) -> Optional[bytes]:
length = self.headers.get("Content-Length")
if not length:
return None
n = int(length)
if n > 16 * 1024 * 1024:
raise ValueError("request body too large for this pull-only POC")
return self.rfile.read(n)
def _forward_streaming(self, target_url: str, public_netloc: str, client_ip: str, stats: Optional[StreamStats] = None) -> None:
cfg = CONFIG
assert cfg is not None
method = self.command
body = None if method in ("GET", "HEAD", "OPTIONS") else self._read_small_body()
redirects = 0
current_url = target_url
drop_auth = False
while True:
p = urlsplit(current_url)
if p.scheme not in ("http", "https") or not p.netloc:
raise ValueError(f"bad upstream URL: {current_url}")
headers = copy_request_headers(self.headers, p.netloc, drop_authorization=drop_auth)
ensure_manifest_accept(headers, p.path or "/")
add_forwarded_for(headers, client_ip)
path = urlunsplit(("", "", p.path or "/", p.query, ""))
conn_cls = http.client.HTTPSConnection if p.scheme == "https" else http.client.HTTPConnection
conn = conn_cls(p.hostname, port=p.port, timeout=cfg.upstream_timeout)
try:
conn.request(method, path, body=body, headers=headers)
resp = conn.getresponse()
if stats is not None:
stats.response_status = resp.status
if resp.status in (301, 302, 303, 307, 308) and resp.getheader("Location"):
if redirects >= MAX_REDIRECTS:
raise RuntimeError("too many upstream redirects")
location = resp.getheader("Location")
resp.close()
conn.close()
new_url = urljoin(current_url, location)
old_host = urlsplit(current_url).netloc.lower()
new_host = urlsplit(new_url).netloc.lower()
if new_host != old_host:
drop_auth = True
current_url = new_url
redirects += 1
continue
self.send_response(resp.status, resp.reason)
has_content_length = False
for k, v in resp.getheaders():
lk = k.lower()
if lk in HOP_BY_HOP_HEADERS:
continue
if lk == "www-authenticate":
v = rewrite_www_authenticate(v, public_netloc)
if lk == "content-length":
has_content_length = True
self.send_header(k, v)
if not has_content_length:
self.close_connection = True
self.end_headers()
if method != "HEAD":
while True:
chunk = resp.read(COPY_CHUNK_SIZE)
if not chunk:
break
if stats is not None:
stats.body_bytes += len(chunk)
self.wfile.write(chunk)
resp.close()
conn.close()
return
except Exception:
conn.close()
raise
def parse_upstreams(items: Iterable[str], base_domain: str) -> Dict[str, Upstream]:
mapping = dict(DEFAULT_UPSTREAMS)
for item in items:
if "=" not in item:
die(f"bad --upstream {item!r}; expected alias=https://registry.example.com")
alias_raw, url_raw = item.split("=", 1)
alias = validate_upstream_alias(alias_raw)
url = url_raw.strip().rstrip("/")
p = urlsplit(url)
if p.scheme not in ("http", "https") or not p.netloc or p.query or p.fragment:
die(f"bad upstream URL {url!r}; expected http(s)://registry.example.com")
mapping[alias] = url
result: Dict[str, Upstream] = {}
for alias, url in mapping.items():
public_host = f"{alias}.{base_domain}".lower()
result[public_host] = Upstream(alias=alias, public_host=public_host, registry_base=url)
return result
def run_certbot(cmd: List[str]) -> None:
log("running: " + " ".join(shlex.quote(x) for x in cmd))
subprocess.run(cmd, check=True)
def is_staging_cert(cert_file: str, cert_name: str) -> bool:
markers = ("STAGING", "Fake LE", "Fake ISRG", "Pebble")
try:
decoded = ssl._ssl._test_decode_cert(cert_file) # type: ignore[attr-defined]
issuer_parts = []
for rdn in decoded.get("issuer", ()): # tuple(tuple(key, value))
for k, v in rdn:
issuer_parts.append(f"{k}={v}")
issuer = ", ".join(issuer_parts)
if any(m.lower() in issuer.lower() for m in markers):
return True
except Exception:
pass
renewal_conf = f"/etc/letsencrypt/renewal/{cert_name}.conf"
try:
with open(renewal_conf, "r", encoding="utf-8", errors="ignore") as f:
data = f.read().lower()
if "acme-staging" in data or "--staging" in data or "test-cert" in data:
return True
except FileNotFoundError:
pass
except Exception:
pass
return False
def delete_certbot_cert(cert_name: str) -> None:
run_certbot(["certbot", "delete", "--cert-name", cert_name, "--non-interactive"])
def ensure_certbot_cert(args: argparse.Namespace, domains: List[str], cert_name: str, cert_file: str) -> None:
if shutil.which("certbot") is None:
die("certbot not found; install it first, e.g. apt install certbot")
os.makedirs(args.acme_webroot, mode=0o755, exist_ok=True)
if os.path.exists(cert_file) and not args.staging and is_staging_cert(cert_file, cert_name):
log(f"existing certificate {cert_name!r} looks like a Let's Encrypt staging/test certificate; deleting before requesting production cert")
delete_certbot_cert(cert_name)
cmd = [
"certbot",
"certonly",
"--webroot",
"-w",
args.acme_webroot,
"--non-interactive",
"--agree-tos",
"--cert-name",
cert_name,
"--keep-until-expiring",
"--expand",
]
if args.email:
cmd.extend(["--email", args.email])
else:
cmd.append("--register-unsafely-without-email")
if args.staging:
cmd.append("--staging")
for d in domains:
cmd.extend(["-d", d])
run_certbot(cmd)
def renewal_loop(interval_seconds: int, cert_file: str, key_file: str) -> None:
global TLS_CONTEXT
while not STOP_EVENT.wait(interval_seconds):
try:
run_certbot(["certbot", "renew", "--quiet"])
if TLS_CONTEXT:
TLS_CONTEXT.load_cert_chain(certfile=cert_file, keyfile=key_file)
log("TLS certificate reloaded")
except Exception as e:
log(f"renewal failed: {type(e).__name__}: {e}")
def register_server(srv: ThreadingHTTPServer, label: str) -> ThreadingHTTPServer:
SERVERS.append(srv)
t = threading.Thread(target=srv.serve_forever, kwargs={"poll_interval": 0.5}, daemon=True, name=label)
SERVER_THREADS.append(t)
t.start()
return srv
def start_http_challenge_server(bind: str, port: int) -> ThreadingHTTPServer:
srv = ReuseThreadingHTTPServer((bind, port), ChallengeAndRedirectHandler)
register_server(srv, f"http-challenge-{port}")
log(f"HTTP challenge/redirect server listening on {bind}:{port}")
return srv
def start_http_proxy_server(bind: str, port: int) -> ThreadingHTTPServer:
srv = ReuseThreadingHTTPServer((bind, port), OCIProxyHandler)
register_server(srv, f"http-proxy-{port}")
log(f"HTTP OCI proxy listening on {bind}:{port}")
return srv
def start_https_proxy_server(bind: str, port: int, cert_file: str, key_file: str) -> ThreadingHTTPServer:
global TLS_CONTEXT
ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
ctx.minimum_version = ssl.TLSVersion.TLSv1_2
ctx.load_cert_chain(certfile=cert_file, keyfile=key_file)
TLS_CONTEXT = ctx
srv = ReuseThreadingHTTPServer((bind, port), OCIProxyHandler, ssl_context=ctx)
register_server(srv, f"https-proxy-{port}")
log(f"HTTPS OCI proxy listening on {bind}:{port}")
return srv
INSTALL_TARGET = Path("/opt/oci_stream_mirror.py")
SYSTEMD_DIR = Path("/etc/systemd/system")
def _require_root(action: str) -> None:
if os.geteuid() != 0:
raise SystemExit(f"[{action}] please run with sudo/root")
def _validate_service_name(service_name: str) -> str:
name = service_name.strip()
if not name:
raise SystemExit("[service] empty --service-name is not allowed")
if "/" in name or name.endswith(".service"):
raise SystemExit("[service] --service-name should be a bare name, e.g. oci-stream-mirror")
if not re.fullmatch(r"[A-Za-z0-9_.@-]+", name):
raise SystemExit("[service] --service-name contains unsupported characters")
return name
def _strip_installer_args(argv: List[str]) -> List[str]:
out: List[str] = []
i = 0
while i < len(argv):
arg = argv[i]
if arg in {"--install", "--uninstall"}:
i += 1
continue
if arg == "--service-name":
i += 2
continue
if arg.startswith("--service-name="):
i += 1
continue
out.append(arg)
i += 1
return out
def _self_script_path() -> Path:
try:
src = Path(__file__).resolve()
except NameError:
raise SystemExit("[install] cannot locate current script path; run from a real .py file")
if not src.exists() or not src.is_file():
raise SystemExit(f"[install] current script is not a regular file: {src}")
return src
def _quote_exec_args(argv: List[str]) -> str:
return " ".join(shlex.quote(x) for x in argv)
def install_systemd_service(args: argparse.Namespace, original_argv: List[str]) -> None:
_require_root("install")
service_name = _validate_service_name(args.service_name)
src = _self_script_path()
dst = INSTALL_TARGET
unit_path = SYSTEMD_DIR / f"{service_name}.service"
runtime_argv = _strip_installer_args(original_argv)
python_bin = shutil.which("python3") or sys.executable or "/usr/bin/python3"
exec_start = f"{shlex.quote(python_bin)} {shlex.quote(str(dst))}"
if runtime_argv:
exec_start += " " + _quote_exec_args(runtime_argv)
print(f"[install] copying {src} -> {dst}")
dst.parent.mkdir(parents=True, exist_ok=True)
if src != dst:
shutil.copy2(src, dst)
dst.chmod(0o755)
unit = textwrap.dedent(f"""\
[Unit]
Description=OCI Streaming Mirror
Wants=network-online.target
After=network-online.target
[Service]
Type=simple
User=root
WorkingDirectory=/opt
Environment=PYTHONUNBUFFERED=1
ExecStart={exec_start}
Restart=on-failure
RestartSec=3
KillSignal=SIGINT
TimeoutStopSec=20
LimitNOFILE=1048576
[Install]
WantedBy=multi-user.target
""")
print(f"[install] writing {unit_path}")
unit_path.write_text(unit, encoding="utf-8")
print("[install] running systemctl daemon-reload")
subprocess.run(["systemctl", "daemon-reload"], check=True)
print()
print("[install] installed successfully.")
print()
print("Start the service:")
print(f" sudo systemctl enable --now {service_name}")
print()
print("View logs:")
print(f" sudo journalctl -u {service_name} -f")
print()
print("Restart after editing arguments or script:")
print(f" sudo systemctl restart {service_name}")
print()
print("Uninstall:")
print(f" sudo python3 {dst} --uninstall --service-name {service_name}")
print()
print("Generated ExecStart:")
print(f" {exec_start}")
def uninstall_systemd_service(service_name_raw: str) -> None:
_require_root("uninstall")
service_name = _validate_service_name(service_name_raw)
unit_path = SYSTEMD_DIR / f"{service_name}.service"
print(f"[uninstall] stopping/disabling {service_name}.service if present")
subprocess.run(["systemctl", "disable", "--now", service_name], check=False)
if unit_path.exists():
print(f"[uninstall] removing {unit_path}")
unit_path.unlink()
else:
print(f"[uninstall] unit not found: {unit_path}")
if INSTALL_TARGET.exists():
print(f"[uninstall] removing {INSTALL_TARGET}")
INSTALL_TARGET.unlink()
else:
print(f"[uninstall] script not found: {INSTALL_TARGET}")
print("[uninstall] running systemctl daemon-reload")
subprocess.run(["systemctl", "daemon-reload"], check=False)
subprocess.run(["systemctl", "reset-failed", service_name], check=False)
print("[uninstall] done")
def build_arg_parser() -> argparse.ArgumentParser:
p = argparse.ArgumentParser(description="No-cache streaming OCI/Docker registry reverse proxy POC")
p.add_argument("--install", action="store_true", help="install this script as a systemd service and exit")
p.add_argument("--uninstall", action="store_true", help="remove the systemd service and /opt/oci_stream_mirror.py, then exit")
p.add_argument("--service-name", default="oci-stream-mirror", help="systemd service name used by --install/--uninstall")
p.add_argument("--base-domain", help="base domain, e.g. mirrors.example.com")
p.add_argument("--email", help="email for Let's Encrypt / certbot; omitted => --register-unsafely-without-email")
p.add_argument("--bind", default="0.0.0.0")
p.add_argument("--http-port", type=int, default=80)
p.add_argument("--https-port", type=int, default=443)
p.add_argument("--http-only", action="store_true", help="serve OCI proxy as plain HTTP; use behind HTTPS reverse proxy or as an insecure registry")
p.add_argument("--external-scheme", choices=["https", "http"], default="https", help="scheme advertised in WWW-Authenticate realm; use http only for direct insecure-registry testing")
p.add_argument("--no-http-server", action="store_true", help="do not bind the built-in HTTP challenge/redirect server; use with --acme-webroot served by an existing port-80 webserver")
p.add_argument("--acme-webroot", default="/var/lib/oci-stream-mirror/acme")
p.add_argument("--cert-name", help="certbot certificate name; default is first configured host")
p.add_argument("--cert-file", help="use existing certificate file instead of /etc/letsencrypt/live/<cert-name>/fullchain.pem")
p.add_argument("--key-file", help="use existing private key file instead of /etc/letsencrypt/live/<cert-name>/privkey.pem")
p.add_argument("--skip-certbot", action="store_true", help="do not call certbot; use existing cert files")
p.add_argument("--staging", action="store_true", help="use Let's Encrypt staging CA for testing")
p.add_argument("--renew-loop", action="store_true", help="run certbot renew once per week and reload cert in memory")
p.add_argument("--allow-cidr", action="append", default=[], help="allow client CIDR; repeatable. Empty means allow all")
p.add_argument("--trust-forwarded-for", action="store_true", help="trust X-Forwarded-For/X-Real-IP from a local reverse proxy for allow-cidr and bans")
p.add_argument("--path-auth", metavar="TOKEN", help="enable path-token auth: ghcr.<base>/<token>/owner/image:tag")
p.add_argument("--serve-base-domain", action="store_true", help="serve an informational page on the base domain itself; requires DNS for the base domain")
p.add_argument("--web-root", help="serve this static directory on the base domain instead of the built-in informational page; implies --serve-base-domain")
p.add_argument("--ban-threshold", type=int, default=10, help="bad path-token attempts before temporary ban")
p.add_argument("--ban-seconds", type=int, default=3600, help="temporary ban duration after too many bad path-token attempts")
p.add_argument("--upstream", action="append", default=[], help="add/override upstream, e.g. --upstream ghcr=https://ghcr.io")
p.add_argument("--upstream-timeout", type=int, default=60, help="per upstream socket operation timeout in seconds")
p.add_argument("--client-timeout", type=int, default=300, help="per client socket read/write idle timeout in seconds")
p.add_argument("--tls-handshake-timeout", type=int, default=10, help="TLS handshake timeout in seconds; protects against handshake slowloris")
p.add_argument("--max-clients", type=int, default=256, help="maximum concurrent client handler threads per listening server")
p.add_argument("--pull-log-idle", type=float, default=DEFAULT_PULL_LOG_IDLE_SECONDS, help="seconds of pull inactivity before printing the PULL END summary")
p.add_argument("--verbose", action="store_true")
return p
def shutdown_all() -> None:
log("shutting down")
STOP_EVENT.set()
PULL_LOG.finalize_all()
shutdown_threads: List[threading.Thread] = []
for srv in list(SERVERS):
t = threading.Thread(target=srv.shutdown, daemon=True)
t.start()
shutdown_threads.append(t)
for t in shutdown_threads:
t.join(timeout=3.0)
for srv in list(SERVERS):
try:
srv.server_close()
except Exception:
pass
for t in list(SERVER_THREADS):
t.join(timeout=2.0)
def main() -> None:
global CONFIG
args = build_arg_parser().parse_args()
if args.install and args.uninstall:
die("--install and --uninstall cannot be used together")
if args.uninstall:
uninstall_systemd_service(args.service_name)
return
if not args.base_domain:
die("--base-domain is required unless --uninstall is used")
base_domain = validate_dns_name(args.base_domain, "--base-domain")
http_port = validate_port(args.http_port, "--http-port")
https_port = validate_port(args.https_port, "--https-port")
upstreams = parse_upstreams(args.upstream, base_domain)
serve_base_domain = bool(args.serve_base_domain or args.web_root)
domains = sorted(upstreams.keys())
if serve_base_domain and base_domain not in domains:
domains.append(base_domain)
domains.sort()
cert_name = validate_cert_name(args.cert_name) if args.cert_name else domains[0]
cert_file = args.cert_file or f"/etc/letsencrypt/live/{cert_name}/fullchain.pem"
key_file = args.key_file or f"/etc/letsencrypt/live/{cert_name}/privkey.pem"
allowed_cidrs = []
for item in args.allow_cidr:
try:
allowed_cidrs.append(ipaddress.ip_network(item, strict=False))
except ValueError as e:
die(f"bad --allow-cidr {item!r}: {e}")
web_root = None
if args.web_root:
web_root_path = Path(args.web_root).resolve()
if not web_root_path.exists() or not web_root_path.is_dir():
die("--web-root must point to an existing directory")
web_root = str(web_root_path)
path_auth_enabled = bool(args.path_auth)
path_token = args.path_auth
if path_token and not SAFE_TOKEN_RE.fullmatch(path_token):
die("--path-auth TOKEN must match [A-Za-z0-9._~-]{1,64}")
if args.ban_threshold <= 0:
die("--ban-threshold must be positive")
if args.ban_seconds <= 0:
die("--ban-seconds must be positive")
if args.upstream_timeout <= 0:
die("--upstream-timeout must be positive")
if args.client_timeout <= 0:
die("--client-timeout must be positive")
if args.tls_handshake_timeout <= 0:
die("--tls-handshake-timeout must be positive")
if args.max_clients <= 0:
die("--max-clients must be positive")
if args.pull_log_idle <= 0:
die("--pull-log-idle must be positive")
# --install is intentionally handled after validation, so invalid runtime
# arguments are not embedded into systemd.
if args.install:
install_systemd_service(args, sys.argv[1:])
return
CONFIG = AppConfig(
base_domain=base_domain,
bind=args.bind,
http_port=http_port,
https_port=https_port,
acme_webroot=args.acme_webroot,
cert_name=cert_name,
cert_file=cert_file,
key_file=key_file,
serve_base_domain=serve_base_domain,
web_root=web_root,
upstream_by_host=upstreams,
allowed_cidrs=allowed_cidrs,
upstream_timeout=args.upstream_timeout,
client_timeout=args.client_timeout,
tls_handshake_timeout=args.tls_handshake_timeout,
max_clients=args.max_clients,
pull_log_idle=args.pull_log_idle,
verbose=args.verbose,
http_only=args.http_only,
external_scheme=args.external_scheme,
trust_forwarded_for=args.trust_forwarded_for,
path_auth_enabled=path_auth_enabled,
path_token=path_token,
auth_relay_secret=secrets.token_bytes(32),
ban_threshold=args.ban_threshold,
ban_seconds=args.ban_seconds,
)
log("configured mirror hosts:")
for host, up in sorted(upstreams.items()):
log(f" {host} -> {up.registry_base}")
if path_auth_enabled:
log(f"path-token auth enabled; use image names like ghcr.{base_domain}/<TOKEN>/owner/image:tag")
log(f"bad path-token attempts: ban after {args.ban_threshold}, duration {args.ban_seconds}s")
if serve_base_domain:
log(f"base-domain page enabled on {base_domain}; ensure DNS has both {base_domain} and *.{base_domain} pointing here")
if web_root:
log(f"serving base-domain static files from {web_root}")
if allowed_cidrs:
log("allowed CIDRs: " + ", ".join(str(x) for x in allowed_cidrs))
log(f"timeouts: upstream={args.upstream_timeout}s, client_idle={args.client_timeout}s, tls_handshake={args.tls_handshake_timeout}s, max_clients={args.max_clients}, pull_log_idle={args.pull_log_idle}s")
if args.trust_forwarded_for:
log("trusting X-Forwarded-For/X-Real-IP headers")
def _on_signal(signum, frame):
STOP_EVENT.set()
signal.signal(signal.SIGTERM, _on_signal)
signal.signal(signal.SIGINT, _on_signal)
if args.http_only:
if not args.skip_certbot:
log("--http-only: skipping built-in certbot/TLS; terminate HTTPS in your reverse proxy or configure Docker insecure-registries for direct HTTP")
start_http_proxy_server(args.bind, http_port)
else:
if not args.no_http_server:
start_http_challenge_server(args.bind, http_port)
else:
log("built-in HTTP challenge/redirect server disabled; certbot webroot must be served by an existing external port-80 webserver")
if not args.skip_certbot:
ensure_certbot_cert(args, domains, cert_name, cert_file)
if not os.path.exists(cert_file) or not os.path.exists(key_file):
die(f"certificate files not found: {cert_file}, {key_file}")
start_https_proxy_server(args.bind, https_port, cert_file, key_file)
if args.renew_loop and not args.skip_certbot:
threading.Thread(
target=renewal_loop,
args=(7 * 24 * 3600, cert_file, key_file),
daemon=True,
name="certbot-renew-loop",
).start()
try:
while not STOP_EVENT.wait(3600):
pass
finally:
shutdown_all()
if __name__ == "__main__":
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment