macos-tools/harbor/scripts/dockerhub-mirror-auth-proxy.py

#!/usr/bin/env python3
"""Pull-only Docker Hub mirror adapter for Harbor proxy-cache projects.

Docker's registry mirror client expects an unauthenticated registry mirror. A
Harbor proxy-cache project can require a Harbor bearer token even when the
project itself is public. This small local adapter terminates Docker's mirror
request, obtains the Harbor token, and forwards the request to:

  https://$HARBOR_HOST/v2/$HARBOR_PROJECT/<repo>/...

It intentionally supports GET/HEAD only.
"""

from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import json
import os
import ssl
import sys
import time
import urllib.error
import urllib.parse
import urllib.request


HARBOR_HOST = os.environ.get("HARBOR_HOST")
HARBOR_PROJECT = os.environ.get("HARBOR_PROJECT", "dockerhub")
LISTEN = os.environ.get("LISTEN", "0.0.0.0")
PORT = int(os.environ.get("PORT", "5001"))
TOKEN_TTL_SKEW = 120

if not HARBOR_HOST:
    raise SystemExit("HARBOR_HOST is required")

VERIFY_TLS = os.environ.get("HARBOR_VERIFY_TLS", "false").lower() in ("1", "true", "yes")
SSL_CTX = ssl.create_default_context() if VERIFY_TLS else ssl._create_unverified_context()
TOKEN_CACHE = {}


def log(message):
    print(message, flush=True)


def token_for(repo):
    harbor_repo = f"{HARBOR_PROJECT}/{repo}"
    cached = TOKEN_CACHE.get(harbor_repo)
    now = time.time()
    if cached and cached[1] > now + TOKEN_TTL_SKEW:
        return cached[0]

    qs = urllib.parse.urlencode(
        {
            "service": "harbor-registry",
            "scope": f"repository:{harbor_repo}:pull",
        }
    )
    url = f"https://{HARBOR_HOST}/service/token?{qs}"
    with urllib.request.urlopen(url, context=SSL_CTX, timeout=30) as resp:
        payload = json.loads(resp.read().decode("utf-8"))

    token = payload.get("token") or payload.get("access_token")
    if not token:
        raise RuntimeError("Harbor token response did not include token")

    expires_in = int(payload.get("expires_in") or 1800)
    TOKEN_CACHE[harbor_repo] = (token, now + expires_in)
    return token


def split_repo(path):
    if not path.startswith("/v2/"):
        return None, None
    rest = path[len("/v2/") :]
    for marker in ("/manifests/", "/blobs/"):
        idx = rest.find(marker)
        if idx >= 0:
            return rest[:idx], rest
    return None, rest


def filtered_headers(headers):
    keep = {
        "accept-ranges",
        "cache-control",
        "content-length",
        "content-range",
        "content-type",
        "docker-content-digest",
        "docker-distribution-api-version",
        "etag",
        "last-modified",
        "location",
    }
    return [(k, v) for k, v in headers.items() if k.lower() in keep]


class ProxyHandler(BaseHTTPRequestHandler):
    server_version = "dockerhub-harbor-auth-proxy/0.2"

    def do_GET(self):
        self.forward()

    def do_HEAD(self):
        self.forward()

    def do_POST(self):
        self.send_error(405, "pull-only proxy")

    def do_PUT(self):
        self.send_error(405, "pull-only proxy")

    def do_PATCH(self):
        self.send_error(405, "pull-only proxy")

    def do_DELETE(self):
        self.send_error(405, "pull-only proxy")

    def forward(self):
        parsed = urllib.parse.urlsplit(self.path)

        if parsed.path in ("/healthz", "/readyz"):
            self.send_response(200)
            self.end_headers()
            if self.command != "HEAD":
                self.wfile.write(b"ok\n")
            return

        if parsed.path in ("/v2", "/v2/"):
            self.send_response(200)
            self.send_header("Docker-Distribution-Api-Version", "registry/2.0")
            self.end_headers()
            return

        repo, rest = split_repo(parsed.path)
        if not repo:
            self.send_error(404, "unsupported registry path")
            return

        query = urllib.parse.parse_qsl(parsed.query, keep_blank_values=True)
        if not any(k == "ns" for k, _ in query):
            query.append(("ns", "docker.io"))
        qs = urllib.parse.urlencode(query)

        upstream = f"https://{HARBOR_HOST}/v2/{HARBOR_PROJECT}/{rest}"
        if qs:
            upstream += f"?{qs}"

        try:
            token = token_for(repo)
            req = urllib.request.Request(upstream, method=self.command)
            req.add_header("Authorization", f"Bearer {token}")
            for name in ("Accept", "Range", "If-None-Match", "If-Modified-Since"):
                value = self.headers.get(name)
                if value:
                    req.add_header(name, value)

            with urllib.request.urlopen(req, context=SSL_CTX, timeout=120) as resp:
                self.send_response(resp.status)
                for k, v in filtered_headers(resp.headers):
                    self.send_header(k, v)
                self.send_header("Docker-Distribution-Api-Version", "registry/2.0")
                self.end_headers()
                if self.command != "HEAD":
                    while True:
                        chunk = resp.read(1024 * 1024)
                        if not chunk:
                            break
                        self.wfile.write(chunk)
            log(f"{self.command} {parsed.path} -> {resp.status} repo={repo}")
        except urllib.error.HTTPError as e:
            body = e.read()
            self.send_response(e.code)
            for k, v in filtered_headers(e.headers):
                self.send_header(k, v)
            self.send_header("Docker-Distribution-Api-Version", "registry/2.0")
            self.end_headers()
            if self.command != "HEAD":
                self.wfile.write(body)
            log(f"{self.command} {parsed.path} -> {e.code} repo={repo}")
        except Exception as e:
            log(f"{self.command} {parsed.path} -> 502 {type(e).__name__}: {e}")
            self.send_error(502, str(e))

    def log_message(self, fmt, *args):
        log("%s - %s" % (self.address_string(), fmt % args))


def main():
    server = ThreadingHTTPServer((LISTEN, PORT), ProxyHandler)
    log(f"listening on {LISTEN}:{PORT}, upstream=https://{HARBOR_HOST}/{HARBOR_PROJECT}")
    server.serve_forever()


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        sys.exit(0)