coreci
da15bda6d2
---ci--- phase: 3 milestone: v0.2 status: complete requirements: covered: [REQ-21, REQ-22, REQ-23] partial: [] ---/ci---
799 lines
36 KiB
Python
799 lines
36 KiB
Python
"""Stub-registry tests for scripts/publish_image.py (REQ-23, G-019).
|
|
|
|
Two local stub servers stand in for the public image registry (anonymous
|
|
pull) and the gitea container registry (JWT auth + blob uploads +
|
|
manifest PUTs). No network access; tiny in-memory fixtures; every
|
|
spawned server is cleaned via addCleanup.
|
|
"""
|
|
|
|
import base64
|
|
import hashlib
|
|
import io
|
|
import json
|
|
import os
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import unittest
|
|
import urllib.parse
|
|
from contextlib import redirect_stderr, redirect_stdout
|
|
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
|
|
|
_SCRIPTS_DIR = os.path.join(os.path.dirname(os.path.dirname(
|
|
os.path.abspath(__file__))), "scripts")
|
|
if _SCRIPTS_DIR not in sys.path:
|
|
sys.path.insert(0, _SCRIPTS_DIR)
|
|
|
|
import publish_image # noqa: E402 (scripts/ dir on sys.path)
|
|
|
|
OCI_MANIFEST = "application/vnd.oci.image.manifest.v1+json"
|
|
OCI_INDEX = "application/vnd.oci.image.index.v1+json"
|
|
OCI_CONFIG = "application/vnd.oci.image.config.v1+json"
|
|
OCI_LAYER = "application/vnd.oci.image.layer.v1.tar+gzip"
|
|
OWNER = "testowner"
|
|
REPO = "testrepo"
|
|
NAME = "{}/{}".format(OWNER, REPO)
|
|
|
|
|
|
def _sha(data):
|
|
return "sha256:" + hashlib.sha256(data).hexdigest()
|
|
|
|
|
|
def _dump(payload):
|
|
return json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()
|
|
|
|
|
|
class HubFixtures:
|
|
"""A tiny synthetic base image: index -> amd64/arm64 children."""
|
|
|
|
def __init__(self):
|
|
self.layer0 = b"base layer zero payload\x00" + b"\xa5" * 64
|
|
self.layer1 = b"base layer one payload" * 7
|
|
self.config = _dump({
|
|
"architecture": "amd64",
|
|
"os": "linux",
|
|
"config": {"Cmd": ["python3"], "Env": ["PATH=/usr/local/bin"]},
|
|
"rootfs": {"type": "layers",
|
|
"diff_ids": [_sha(self.layer0), _sha(self.layer1)]},
|
|
})
|
|
self.child = {
|
|
"schemaVersion": 2,
|
|
"mediaType": OCI_MANIFEST,
|
|
"config": {"mediaType": OCI_CONFIG,
|
|
"digest": _sha(self.config),
|
|
"size": len(self.config)},
|
|
"layers": [
|
|
{"mediaType": OCI_LAYER, "digest": _sha(self.layer0),
|
|
"size": len(self.layer0)},
|
|
{"mediaType": OCI_LAYER, "digest": _sha(self.layer1),
|
|
"size": len(self.layer1)},
|
|
],
|
|
}
|
|
self.child_digest = _sha(_dump(self.child))
|
|
arm64_child = dict(self.child)
|
|
arm64_child["annotations"] = {"test-fixture": "arm64"}
|
|
self.arm64_child_digest = _sha(_dump(arm64_child))
|
|
self.index = {
|
|
"schemaVersion": 2,
|
|
"mediaType": OCI_INDEX,
|
|
"manifests": [
|
|
{"mediaType": OCI_MANIFEST, "digest": self.child_digest,
|
|
"size": len(_dump(self.child)),
|
|
"platform": {"os": "linux", "architecture": "amd64"}},
|
|
{"mediaType": OCI_MANIFEST,
|
|
"digest": self.arm64_child_digest,
|
|
"size": len(_dump(arm64_child)),
|
|
"platform": {"os": "linux", "architecture": "arm64"}},
|
|
],
|
|
}
|
|
self.blobs = {
|
|
_sha(self.config): self.config,
|
|
_sha(self.layer0): self.layer0,
|
|
_sha(self.layer1): self.layer1,
|
|
}
|
|
|
|
|
|
class _StubHubHandler(BaseHTTPRequestHandler):
|
|
fixtures = None # set per test class run
|
|
requests = None
|
|
token_mode = "ok" # ok | malformed (200 with an INVALID JSON body)
|
|
|
|
def _record(self):
|
|
length = int(self.headers.get("Content-Length") or 0)
|
|
body = self.rfile.read(length) if length else b""
|
|
self.requests.append({
|
|
"method": self.command,
|
|
"path": self.path,
|
|
"authorization": self.headers.get("Authorization", ""),
|
|
"accept": self.headers.get("Accept", ""),
|
|
"body": body,
|
|
})
|
|
|
|
def _send(self, status, payload, content_type):
|
|
data = payload if isinstance(payload, bytes) else _dump(payload)
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", content_type)
|
|
self.send_header("Content-Length", str(len(data)))
|
|
self.end_headers()
|
|
if self.command != "HEAD":
|
|
self.wfile.write(data)
|
|
|
|
def do_GET(self):
|
|
self._record()
|
|
path = urllib.parse.urlsplit(self.path).path
|
|
query = dict(urllib.parse.parse_qsl(
|
|
urllib.parse.urlsplit(self.path).query))
|
|
if path == "/token":
|
|
if self.token_mode == "malformed":
|
|
# 200 + invalid JSON: the CWE-754 probe — must surface as
|
|
# a remote failure WITH the IMAGE_PENDING marker, never
|
|
# a bare traceback.
|
|
self._send(200, b"this { is not valid json", "application/json")
|
|
else:
|
|
self._send(200, {"token": "hub-anon-token"}, "application/json")
|
|
elif path == "/v2/library/python/manifests/3.11-slim":
|
|
self._send(200, self.fixtures.index, OCI_INDEX)
|
|
elif path == "/v2/library/python/manifests/" + self.fixtures.child_digest:
|
|
self._send(200, self.fixtures.child, OCI_MANIFEST)
|
|
else:
|
|
digest = path.rsplit("/", 1)[-1]
|
|
if digest in self.fixtures.blobs:
|
|
self._send(200, self.fixtures.blobs[digest],
|
|
"application/octet-stream")
|
|
else:
|
|
self._send(404, {"errors": [{"digest": digest}]},
|
|
"application/json")
|
|
|
|
def log_message(self, format, *args):
|
|
pass
|
|
|
|
|
|
class _StubGiteaHandler(BaseHTTPRequestHandler):
|
|
"""Scriptable gitea registry stub.
|
|
|
|
state dict keys: requests, uploaded, token_mode (ok|empty),
|
|
upload_mode (monolithic|redirect307|redirect307_cross|stall),
|
|
manifest_mode (ok|fail500), redirect_seen, manifest_puts,
|
|
cross_origin_base (absolute URL of the second-origin stub used by
|
|
redirect307_cross).
|
|
"""
|
|
|
|
state = None
|
|
|
|
def _record(self):
|
|
length = int(self.headers.get("Content-Length") or 0)
|
|
body = self.rfile.read(length) if length else b""
|
|
self.state["requests"].append({
|
|
"method": self.command,
|
|
"path": self.path,
|
|
"authorization": self.headers.get("Authorization", ""),
|
|
"content_type": self.headers.get("Content-Type", ""),
|
|
"body": body,
|
|
})
|
|
|
|
def _send(self, status, payload=b"", content_type="application/json",
|
|
headers=None):
|
|
data = payload if isinstance(payload, bytes) else _dump(payload)
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", content_type)
|
|
self.send_header("Content-Length", str(len(data)))
|
|
for key, value in (headers or {}).items():
|
|
self.send_header(key, value)
|
|
self.end_headers()
|
|
if self.command != "HEAD":
|
|
self.wfile.write(data)
|
|
|
|
def do_GET(self):
|
|
self._record()
|
|
parsed = urllib.parse.urlsplit(self.path)
|
|
if parsed.path == "/v2/":
|
|
# G-013 smoke check 1: the registry auth challenge (401 +
|
|
# Bearer; the realm defaults to this stub's own token
|
|
# endpoint so pre-existing state dicts work unchanged).
|
|
realm = self.state.get("challenge_realm") or \
|
|
"http://{}:{}/v2/token".format(
|
|
self.server.server_address[0],
|
|
self.server.server_address[1])
|
|
self._send(401, {"errors": [{"code": "UNAUTHORIZED"}]},
|
|
headers={
|
|
"WWW-Authenticate":
|
|
'Bearer realm="{}",service="{}"'.format(
|
|
realm, publish_image.GITEA_AUTH_SERVICE)})
|
|
return
|
|
if parsed.path == "/v2/token":
|
|
if self.state["token_mode"] == "empty":
|
|
self._send(200, {"token": ""})
|
|
else:
|
|
self._send(200, {"token": "stub-jwt"})
|
|
return
|
|
if parsed.path.startswith("/api/v1/repos/") and \
|
|
"/releases/tags/" in parsed.path:
|
|
# G-013 smoke check 3: the gitea API release-by-tag GET
|
|
# (state["releases"]: tag -> release payload; absent -> 404).
|
|
tag = parsed.path.rsplit("/", 1)[-1]
|
|
release = self.state.get("releases", {}).get(tag)
|
|
if release is None:
|
|
self._send(404, {"errors": [{"message": "release not found"}]})
|
|
else:
|
|
self._send(200, release)
|
|
return
|
|
|
|
def do_HEAD(self):
|
|
self._record()
|
|
parsed = urllib.parse.urlsplit(self.path)
|
|
digest = parsed.path.rsplit("/", 1)[-1]
|
|
if digest in self.state["uploaded"]:
|
|
self._send(200, b"")
|
|
else:
|
|
self._send(404, b"")
|
|
|
|
def do_POST(self):
|
|
self._record()
|
|
parsed = urllib.parse.urlsplit(self.path)
|
|
query = dict(urllib.parse.parse_qsl(parsed.query))
|
|
digest = query.get("digest", "")
|
|
if self.state["upload_mode"] == "stall":
|
|
time.sleep(15)
|
|
return
|
|
if self.state["upload_mode"] == "redirect307_cross" and \
|
|
digest not in self.state["redirect_seen"]:
|
|
# CWE-200 probe: ABSOLUTE Location pointing at a DIFFERENT
|
|
# origin (the second stub server) — the client must follow
|
|
# WITHOUT the Authorization header.
|
|
self.state["redirect_seen"].add(digest)
|
|
location = "{}/v2/{}/blobs/uploads/?digest={}".format(
|
|
self.state["cross_origin_base"], NAME, digest)
|
|
self._send(307, b"", headers={"Location": location})
|
|
return
|
|
if self.state["upload_mode"] == "redirect307" and \
|
|
digest not in self.state["redirect_seen"]:
|
|
self.state["redirect_seen"].add(digest)
|
|
location = "/v2/{}/blobs/uploads/?digest={}".format(NAME, digest)
|
|
self._send(307, b"", headers={"Location": location})
|
|
return
|
|
if _sha(self.state["requests"][-1]["body"]) == digest:
|
|
self.state["uploaded"][digest] = self.state["requests"][-1]["body"]
|
|
self._send(201, {"status": "created"})
|
|
else:
|
|
self._send(400, {"errors": [{"digest": digest}]})
|
|
|
|
def do_PUT(self):
|
|
self._record()
|
|
parsed = urllib.parse.urlsplit(self.path)
|
|
query = dict(urllib.parse.parse_qsl(parsed.query))
|
|
body = self.state["requests"][-1]["body"]
|
|
if "/manifests/" in parsed.path:
|
|
if self.state["manifest_mode"] == "fail500":
|
|
self.state.setdefault("manifest_fail_count", 0)
|
|
self.state["manifest_fail_count"] += 1
|
|
self._send(500, {"errors": [{"code": "UNKNOWN"}]})
|
|
return
|
|
manifest = json.loads(body)
|
|
referenced = [manifest["config"]["digest"]] + [
|
|
layer["digest"] for layer in manifest["layers"]]
|
|
if any(digest not in self.state["uploaded"]
|
|
for digest in referenced):
|
|
self._send(400, {"errors": [{"code": "BLOB_UNKNOWN"}]})
|
|
return
|
|
if self.headers.get("Content-Type") != OCI_MANIFEST:
|
|
self._send(400, {"errors": [{"code": "MANIFEST_INVALID"}]})
|
|
return
|
|
self.state.setdefault("manifest_puts", []).append(
|
|
{"tag": parsed.path.rsplit("/", 1)[-1],
|
|
"content_type": self.headers.get("Content-Type", ""),
|
|
"manifest": manifest})
|
|
self._send(201, {})
|
|
return
|
|
digest = query.get("digest", "")
|
|
if _sha(body) == digest:
|
|
self.state["uploaded"][digest] = body
|
|
self._send(201, {})
|
|
else:
|
|
self._send(400, {"errors": [{"code": "DIGEST_INVALID"}]})
|
|
|
|
|
|
class _StubCrossOriginHandler(BaseHTTPRequestHandler):
|
|
"""Second-origin stub (another port) for the CWE-200 redirect probe.
|
|
|
|
state dict keys: requests, uploaded — ``uploaded`` is SHARED with the
|
|
gitea stub (the redirect target models the registry's presigned
|
|
external storage; a landed blob is visible to the registry), while
|
|
``requests`` stays separate so the test can assert on exactly what
|
|
this origin received — above all that NO Authorization header
|
|
crossed the origin boundary.
|
|
"""
|
|
|
|
state = None
|
|
|
|
def _record(self):
|
|
length = int(self.headers.get("Content-Length") or 0)
|
|
body = self.rfile.read(length) if length else b""
|
|
self.state["requests"].append({
|
|
"method": self.command,
|
|
"path": self.path,
|
|
"authorization": self.headers.get("Authorization", ""),
|
|
"content_type": self.headers.get("Content-Type", ""),
|
|
"body": body,
|
|
})
|
|
return body
|
|
|
|
def _send(self, status, payload=b"", content_type="application/json"):
|
|
data = payload if isinstance(payload, bytes) else _dump(payload)
|
|
self.send_response(status)
|
|
self.send_header("Content-Type", content_type)
|
|
self.send_header("Content-Length", str(len(data)))
|
|
self.end_headers()
|
|
if self.command != "HEAD":
|
|
self.wfile.write(data)
|
|
|
|
def do_POST(self):
|
|
body = self._record()
|
|
query = dict(urllib.parse.parse_qsl(
|
|
urllib.parse.urlsplit(self.path).query))
|
|
digest = query.get("digest", "")
|
|
if _sha(body) == digest:
|
|
self.state["uploaded"][digest] = body
|
|
self._send(201, {"status": "created"})
|
|
else:
|
|
self._send(400, {"errors": [{"code": "DIGEST_INVALID"}]})
|
|
|
|
def log_message(self, format, *args):
|
|
pass
|
|
|
|
|
|
class PublishImageTests(unittest.TestCase):
|
|
"""End-to-end publish flows against local stub registries."""
|
|
|
|
TAG = "v0.2.2-test"
|
|
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.fixtures = HubFixtures()
|
|
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.artifact = os.path.join(self.temp.name, "coreci-broker.pyz")
|
|
with open(self.artifact, "wb") as handle:
|
|
handle.write(b"fake pyz payload for tests" * 3)
|
|
self.env_root = os.path.join(self.temp.name, "envroot")
|
|
os.makedirs(self.env_root)
|
|
with open(os.path.join(self.env_root, ".env.secrets"), "w") as handle:
|
|
handle.write("GITEA_TOKEN=testtoken123\n")
|
|
|
|
self.hub_requests = []
|
|
_StubHubHandler.fixtures = self.fixtures
|
|
_StubHubHandler.requests = self.hub_requests
|
|
_StubHubHandler.token_mode = "ok"
|
|
self.hub = ThreadingHTTPServer(("127.0.0.1", 0), _StubHubHandler)
|
|
threading.Thread(target=self.hub.serve_forever, daemon=True).start()
|
|
self.addCleanup(self.hub.shutdown)
|
|
self.addCleanup(self.hub.server_close)
|
|
|
|
self.gitea_state = {
|
|
"requests": [], "uploaded": {}, "token_mode": "ok",
|
|
"upload_mode": "monolithic", "manifest_mode": "ok",
|
|
"redirect_seen": set(),
|
|
}
|
|
_StubGiteaHandler.state = self.gitea_state
|
|
self.gitea = ThreadingHTTPServer(("127.0.0.1", 0), _StubGiteaHandler)
|
|
threading.Thread(target=self.gitea.serve_forever, daemon=True).start()
|
|
self.addCleanup(self.gitea.shutdown)
|
|
self.addCleanup(self.gitea.server_close)
|
|
|
|
def run_publish(self, tag=None, timeout=60, artifact=None, env_root=None):
|
|
argv = [
|
|
tag or self.TAG,
|
|
"--artifact", artifact or self.artifact,
|
|
"--hub-registry", "http://127.0.0.1:{}".format(self.hub.server_address[1]),
|
|
"--hub-auth", "http://127.0.0.1:{}/token".format(self.hub.server_address[1]),
|
|
"--gitea-registry", "http://127.0.0.1:{}".format(self.gitea.server_address[1]),
|
|
"--owner", OWNER, "--repo", REPO,
|
|
"--env-root", env_root or self.env_root,
|
|
"--timeout", str(timeout),
|
|
]
|
|
out, err = io.StringIO(), io.StringIO()
|
|
with redirect_stdout(out), redirect_stderr(err):
|
|
code = publish_image.main(argv)
|
|
return code, out.getvalue(), err.getvalue()
|
|
|
|
# 1 -----------------------------------------------------------------
|
|
def test_happy_path_end_to_end(self):
|
|
code, out, err = self.run_publish()
|
|
self.assertEqual(code, 0, err or out)
|
|
puts = self.gitea_state.get("manifest_puts", [])
|
|
self.assertEqual([entry["tag"] for entry in puts],
|
|
[self.TAG, "latest"],
|
|
"version tag FIRST, then latest (D-021)")
|
|
manifest = puts[0]["manifest"]
|
|
self.assertEqual(puts[0]["content_type"], OCI_MANIFEST)
|
|
self.assertEqual(manifest["config"]["mediaType"], OCI_CONFIG)
|
|
broker_layer = manifest["layers"][-1]
|
|
self.assertEqual(broker_layer["mediaType"], OCI_LAYER)
|
|
base_descriptors = self.fixtures.child["layers"]
|
|
self.assertEqual(manifest["layers"][:-1], base_descriptors,
|
|
"base layer descriptors reused verbatim")
|
|
# deterministic broker layer digest matches an independent build
|
|
with open(self.artifact, "rb") as handle:
|
|
artifact_bytes = handle.read()
|
|
layer_blob, layer_digest, diff_id = publish_image.build_broker_layer(
|
|
artifact_bytes)
|
|
self.assertEqual(broker_layer["digest"], layer_digest)
|
|
self.assertEqual(broker_layer["size"], len(layer_blob))
|
|
# pushed config content
|
|
pushed_config = json.loads(
|
|
self.gitea_state["uploaded"][manifest["config"]["digest"]])
|
|
self.assertEqual(pushed_config["config"]["Entrypoint"],
|
|
["python3", "/app/coreci-broker.pyz"])
|
|
self.assertEqual(pushed_config["config"]["Cmd"], [])
|
|
self.assertEqual(pushed_config["config"]["WorkingDir"], "/state")
|
|
self.assertEqual(pushed_config["config"]["Labels"],
|
|
{"org.opencontainers.image.version": self.TAG})
|
|
self.assertEqual(pushed_config["rootfs"]["diff_ids"][-1], diff_id)
|
|
self.assertEqual(len(pushed_config["rootfs"]["diff_ids"]),
|
|
len(self.fixtures.child["layers"]) + 1)
|
|
# base config blob uploaded verbatim (digest invariant)
|
|
self.assertEqual(
|
|
self.gitea_state["uploaded"][self.fixtures.child["config"]["digest"]],
|
|
self.fixtures.config)
|
|
# all blobs present
|
|
for descriptor in manifest["layers"]:
|
|
self.assertIn(descriptor["digest"], self.gitea_state["uploaded"])
|
|
self.assertNotIn("IMAGE_PENDING", out)
|
|
|
|
# 2 -----------------------------------------------------------------
|
|
def test_head_skip_blob_not_reuploaded(self):
|
|
preseed = self.fixtures.child["layers"][0]["digest"]
|
|
self.gitea_state["uploaded"][preseed] = self.fixtures.layer0
|
|
code, out, err = self.run_publish()
|
|
self.assertEqual(code, 0, err or out)
|
|
posts = [r for r in self.gitea_state["requests"]
|
|
if r["method"] == "POST" and preseed in r["path"]]
|
|
self.assertEqual(posts, [], "pre-seeded blob must not be re-uploaded")
|
|
hub_blob_gets = [r for r in self.hub_requests
|
|
if r["method"] == "GET" and "/blobs/" in r["path"]
|
|
and preseed in r["path"]]
|
|
self.assertEqual(hub_blob_gets, [],
|
|
"skipped blob is never fetched from the hub")
|
|
|
|
# 3 -----------------------------------------------------------------
|
|
def test_redirect_307_relative_location_followed(self):
|
|
self.gitea_state["upload_mode"] = "redirect307"
|
|
code, out, err = self.run_publish()
|
|
self.assertEqual(code, 0, err or out)
|
|
blob_posts = [r for r in self.gitea_state["requests"]
|
|
if r["method"] == "POST" and "/blobs/uploads/" in r["path"]
|
|
and "digest=" in r["path"]]
|
|
self.assertGreater(len(blob_posts), 0)
|
|
digests = [dict(urllib.parse.parse_qsl(
|
|
urllib.parse.urlsplit(r["path"]).query))["digest"]
|
|
for r in blob_posts]
|
|
self.assertTrue(all(d in self.gitea_state["uploaded"]
|
|
for d in digests),
|
|
"every redirected upload must land")
|
|
# each redirected digest POSTed twice (307 + follow)
|
|
for digest in set(digests):
|
|
self.assertEqual(digests.count(digest), 2,
|
|
"307 then explicit re-POST per digest")
|
|
# same-origin (relative Location) redirects KEEP the
|
|
# Authorization header — the CWE-200 strip is cross-origin ONLY
|
|
for request in blob_posts:
|
|
self.assertEqual(
|
|
request["authorization"], "Bearer stub-jwt",
|
|
"same-origin redirect must keep Authorization")
|
|
|
|
# 4 -----------------------------------------------------------------
|
|
def test_anonymous_token_trap_refused_before_push(self):
|
|
self.gitea_state["token_mode"] = "empty"
|
|
code, out, err = self.run_publish()
|
|
self.assertEqual(code, 1, "remote auth failure -> non-zero")
|
|
self.assertIn("anonymous-200 trap refused", err)
|
|
self.assertIn("IMAGE_PENDING", out)
|
|
post_put = [r for r in self.gitea_state["requests"]
|
|
if r["method"] in ("POST", "PUT")]
|
|
self.assertEqual(post_put, [],
|
|
"no blob upload or manifest PUT may happen")
|
|
|
|
# 5 -----------------------------------------------------------------
|
|
def test_layer_and_config_determinism(self):
|
|
with open(self.artifact, "rb") as handle:
|
|
artifact = handle.read()
|
|
first = publish_image.build_broker_layer(artifact)
|
|
second = publish_image.build_broker_layer(artifact)
|
|
self.assertEqual(first[1], second[1], "layer digest stable")
|
|
self.assertEqual(first[2], second[2], "diff_id stable")
|
|
base_config = json.loads(self.fixtures.config)
|
|
c1 = publish_image.build_config_blob(base_config, first[2], "t1")
|
|
c2 = publish_image.build_config_blob(base_config, first[2], "t1")
|
|
self.assertEqual(_sha(c1), _sha(c2), "config digest stable")
|
|
|
|
# 6 -----------------------------------------------------------------
|
|
def test_amd64_child_selected(self):
|
|
code, out, err = self.run_publish()
|
|
self.assertEqual(code, 0, err or out)
|
|
child_gets = [r for r in self.hub_requests
|
|
if "/manifests/sha256:" in r["path"]]
|
|
self.assertTrue(child_gets, "child manifest must be fetched by digest")
|
|
self.assertTrue(all(self.fixtures.child_digest in r["path"]
|
|
for r in child_gets),
|
|
"amd64 child only")
|
|
self.assertFalse(any(self.fixtures.arm64_child_digest in r["path"]
|
|
for r in self.hub_requests),
|
|
"arm64 child never touched")
|
|
|
|
# 7 -----------------------------------------------------------------
|
|
def test_timeout_aborts_rather_than_hangs(self):
|
|
self.gitea_state["upload_mode"] = "stall"
|
|
started = time.monotonic()
|
|
code, out, err = self.run_publish(timeout=1)
|
|
elapsed = time.monotonic() - started
|
|
self.assertEqual(code, 1)
|
|
self.assertIn("IMAGE_PENDING", out)
|
|
self.assertLess(elapsed, 30, "bounded retries, no hang")
|
|
stalls = [r for r in self.gitea_state["requests"]
|
|
if r["method"] == "POST" and "/blobs/uploads/" in r["path"]]
|
|
self.assertLessEqual(len(stalls), 3, "bounded retry attempts")
|
|
|
|
# 8 -----------------------------------------------------------------
|
|
def test_manifest_retry_exhaustion_marker(self):
|
|
self.gitea_state["manifest_mode"] = "fail500"
|
|
code, out, err = self.run_publish()
|
|
self.assertEqual(code, 1)
|
|
self.assertIn("IMAGE_PENDING", out)
|
|
self.assertEqual(self.gitea_state.get("manifest_fail_count"), 3,
|
|
"exactly three bounded attempts on the tag PUT")
|
|
self.assertNotIn("latest",
|
|
[r["path"].rsplit("/", 1)[-1]
|
|
for r in self.gitea_state["requests"]
|
|
if r["method"] == "PUT"],
|
|
"latest must not be attempted after exhaustion")
|
|
|
|
# 9 -----------------------------------------------------------------
|
|
def test_token_discipline(self):
|
|
code, out, err = self.run_publish()
|
|
self.assertEqual(code, 0, err or out)
|
|
token_calls = [r for r in self.gitea_state["requests"]
|
|
if r["path"].startswith("/v2/token")]
|
|
self.assertTrue(token_calls)
|
|
expected = base64.b64encode(
|
|
b"publisher:testtoken123").decode("ascii")
|
|
for call in token_calls:
|
|
self.assertEqual(call["authorization"],
|
|
"Basic {}".format(expected))
|
|
# scope requested with pull,push (registry v2 scope form)
|
|
query = urllib.parse.unquote(
|
|
urllib.parse.urlsplit(token_calls[0]["path"]).query)
|
|
self.assertIn("repository:{}/{}:pull,push".format(OWNER, REPO),
|
|
query)
|
|
# missing token: local precondition, zero remote requests
|
|
empty_root = os.path.join(self.temp.name, "emptyroot")
|
|
os.makedirs(empty_root)
|
|
before = len(self.gitea_state["requests"]) + len(self.hub_requests)
|
|
code, out, err = self.run_publish(env_root=empty_root)
|
|
self.assertEqual(code, 2, "local precondition -> exit 2")
|
|
self.assertNotIn("IMAGE_PENDING", out, "no marker for local errors")
|
|
after = len(self.gitea_state["requests"]) + len(self.hub_requests)
|
|
self.assertEqual(before, after, "no request of any kind was made")
|
|
|
|
# 10 ----------------------------------------------------------------
|
|
def test_malformed_hub_token_json_keeps_pending_marker(self):
|
|
"""CWE-754 marker discipline: a 200 with an INVALID JSON body from
|
|
the hub token endpoint fails as a remote answer — exit 1 WITH the
|
|
IMAGE_PENDING marker — never an uncaught-ValueError traceback."""
|
|
_StubHubHandler.token_mode = "malformed"
|
|
code, out, err = self.run_publish()
|
|
self.assertEqual(code, 1)
|
|
self.assertIn("malformed JSON", err,
|
|
"the failure class must be named in the error")
|
|
self.assertIn("IMAGE_PENDING", out,
|
|
"marker discipline holds for malformed remote JSON")
|
|
self.assertNotIn("Traceback", out + err)
|
|
self.assertEqual(self.gitea_state["requests"], [],
|
|
"nothing is attempted against gitea")
|
|
|
|
# 11 ----------------------------------------------------------------
|
|
def test_redirect_307_cross_origin_strips_authorization(self):
|
|
"""CWE-200: a 307 whose Location resolves to a DIFFERENT origin
|
|
is followed WITHOUT the Authorization header — the registry JWT
|
|
never crosses origins — and the blob still lands."""
|
|
cross_state = {"requests": [], "uploaded": self.gitea_state["uploaded"]}
|
|
_StubCrossOriginHandler.state = cross_state
|
|
cross = ThreadingHTTPServer(("127.0.0.1", 0), _StubCrossOriginHandler)
|
|
threading.Thread(target=cross.serve_forever, daemon=True).start()
|
|
self.addCleanup(cross.shutdown)
|
|
self.addCleanup(cross.server_close)
|
|
self.gitea_state["upload_mode"] = "redirect307_cross"
|
|
self.gitea_state["cross_origin_base"] = "http://127.0.0.1:{}".format(
|
|
cross.server_address[1])
|
|
code, out, err = self.run_publish()
|
|
self.assertEqual(code, 0, err or out)
|
|
self.assertTrue(cross_state["requests"],
|
|
"the redirect was followed to the other origin")
|
|
for request in cross_state["requests"]:
|
|
self.assertEqual(
|
|
request["authorization"], "",
|
|
"Authorization must be stripped on cross-origin redirects")
|
|
# the blob landed (visible through the shared registry blob store)
|
|
self.assertEqual(len(self.gitea_state["uploaded"]), 5,
|
|
"base config + 2 base layers + broker layer + "
|
|
"image config all landed")
|
|
# and the flow still completes: both manifest PUTs back on the
|
|
# origin, where Authorization remains in play
|
|
self.assertEqual(
|
|
[entry["tag"] for entry in self.gitea_state.get(
|
|
"manifest_puts", [])],
|
|
[self.TAG, "latest"])
|
|
|
|
|
|
class SmokeModeTests(unittest.TestCase):
|
|
"""G-013 --smoke logic, stub-tested.
|
|
|
|
The LIVE smoke never runs in this suite — it executes only at the
|
|
P3-05 ship, where the ship gate sets PUBLISH_SMOKE=1 in the
|
|
environment (the script itself never reads that variable). Here the
|
|
four GET-only checks run against fresh stub instances: zero network,
|
|
and the stub records double as the read-only proof (a POST/PUT
|
|
landing anywhere in a smoke run is a hard failure).
|
|
"""
|
|
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.env_root = os.path.join(self.temp.name, "envroot")
|
|
os.makedirs(self.env_root)
|
|
with open(os.path.join(self.env_root, ".env.secrets"), "w") as handle:
|
|
handle.write("GITEA_TOKEN=testtoken123\n")
|
|
|
|
self.fixtures = HubFixtures()
|
|
self.hub_requests = []
|
|
_StubHubHandler.fixtures = self.fixtures
|
|
_StubHubHandler.requests = self.hub_requests
|
|
_StubHubHandler.token_mode = "ok"
|
|
self.hub = ThreadingHTTPServer(("127.0.0.1", 0), _StubHubHandler)
|
|
threading.Thread(target=self.hub.serve_forever, daemon=True).start()
|
|
self.addCleanup(self.hub.shutdown)
|
|
self.addCleanup(self.hub.server_close)
|
|
|
|
self.gitea_state = {
|
|
"requests": [], "uploaded": {}, "token_mode": "ok",
|
|
"upload_mode": "monolithic", "manifest_mode": "ok",
|
|
"redirect_seen": set(), "releases": {},
|
|
}
|
|
_StubGiteaHandler.state = self.gitea_state
|
|
self.gitea = ThreadingHTTPServer(("127.0.0.1", 0), _StubGiteaHandler)
|
|
threading.Thread(target=self.gitea.serve_forever, daemon=True).start()
|
|
self.addCleanup(self.gitea.shutdown)
|
|
self.addCleanup(self.gitea.server_close)
|
|
# The /v2/ challenge realm must point back at THIS stub's token
|
|
# endpoint so the recorded-realm exchange stays in-stub.
|
|
self.gitea_state["challenge_realm"] = \
|
|
"http://127.0.0.1:{}/v2/token".format(
|
|
self.gitea.server_address[1])
|
|
|
|
def run_smoke(self, tag, env_root=None):
|
|
argv = [
|
|
tag, "--smoke",
|
|
"--hub-registry",
|
|
"http://127.0.0.1:{}".format(self.hub.server_address[1]),
|
|
"--hub-auth",
|
|
"http://127.0.0.1:{}/token".format(
|
|
self.hub.server_address[1]),
|
|
"--gitea-registry",
|
|
"http://127.0.0.1:{}".format(self.gitea.server_address[1]),
|
|
"--owner", OWNER, "--repo", REPO,
|
|
"--env-root", env_root or self.env_root,
|
|
]
|
|
out, err = io.StringIO(), io.StringIO()
|
|
with redirect_stdout(out), redirect_stderr(err):
|
|
code = publish_image.main(argv)
|
|
return code, out.getvalue(), err.getvalue()
|
|
|
|
def _post_put(self):
|
|
"""(gitea POST/PUT, hub POST/PUT) across the stub records."""
|
|
return ([r for r in self.gitea_state["requests"]
|
|
if r["method"] in ("POST", "PUT")],
|
|
[r for r in self.hub_requests
|
|
if r["method"] in ("POST", "PUT")])
|
|
|
|
def _release_gets(self):
|
|
return [r for r in self.gitea_state["requests"]
|
|
if "/releases/tags/" in r["path"]]
|
|
|
|
# 12 ----------------------------------------------------------------
|
|
def test_smoke_all_checks_pass(self):
|
|
self.gitea_state["releases"]["v0.2.0"] = {
|
|
"tag_name": "v0.2.0",
|
|
"assets": [{"name": "phase0-notes.txt"}],
|
|
}
|
|
code, out, err = self.run_smoke("v0.2.0")
|
|
self.assertEqual(code, 0, err or out)
|
|
for marker in ("PASS [1/4]", "PASS [2/4]", "PASS [3/4]",
|
|
"PASS [4/4]"):
|
|
self.assertIn(marker, out)
|
|
self.assertIn("G-021 confirmed: no .pyz asset", out,
|
|
"the v0.2.0 run carries the G-021 forge-side "
|
|
"confirmation")
|
|
self.assertIn("phase0-notes.txt", out, "asset names print as a fact")
|
|
self.assertIn(self.fixtures.child_digest, out,
|
|
"the amd64 child digest prints (drift check)")
|
|
# the manifest GET carries the full Accept set
|
|
manifest_gets = [r for r in self.hub_requests
|
|
if "/manifests/3.11-slim" in r["path"]]
|
|
self.assertEqual(len(manifest_gets), 1)
|
|
self.assertEqual(manifest_gets[0]["accept"],
|
|
publish_image.ACCEPT_MANIFESTS)
|
|
# the release GET uses the gitea API token auth form
|
|
release_gets = self._release_gets()
|
|
self.assertEqual(len(release_gets), 1)
|
|
self.assertEqual(release_gets[0]["authorization"],
|
|
"token testtoken123")
|
|
# READ-ONLY PROOF: zero POST/PUT recorded at either stub
|
|
gitea_writes, hub_writes = self._post_put()
|
|
self.assertEqual(gitea_writes, [], "no gitea write of any kind")
|
|
self.assertEqual(hub_writes, [], "no hub write of any kind")
|
|
self.assertNotIn("IMAGE_PENDING", out)
|
|
|
|
# 13 ----------------------------------------------------------------
|
|
def test_smoke_v0_2_0_no_pyz_assertion(self):
|
|
pyz_release = {"tag_name": "v0.2.0",
|
|
"assets": [{"name": "coreci-broker-v0.2.0.pyz"}]}
|
|
self.gitea_state["releases"]["v0.2.0"] = pyz_release
|
|
self.gitea_state["releases"]["v0.2.2"] = pyz_release
|
|
# v0.2.0 + a .pyz asset: the G-021 hard assertion fails the smoke
|
|
code, out, err = self.run_smoke("v0.2.0")
|
|
self.assertEqual(code, 1)
|
|
self.assertIn("FAILED [3/4]", out)
|
|
self.assertIn("G-021", out, "the failure line names G-021")
|
|
self.assertIn(".pyz", out)
|
|
self.assertEqual(self.hub_requests, [],
|
|
"a FAILED check stops the smoke: the hub "
|
|
"resolve (step 4) never runs")
|
|
# the SAME release under a different tag: the assertion binds
|
|
# v0.2.0 only -> all four checks pass
|
|
code, out, err = self.run_smoke("v0.2.2")
|
|
self.assertEqual(code, 0, err or out)
|
|
for marker in ("PASS [3/4]", "PASS [4/4]"):
|
|
self.assertIn(marker, out)
|
|
self.assertNotIn("G-021 violation", out)
|
|
gitea_writes, hub_writes = self._post_put()
|
|
self.assertEqual(gitea_writes + hub_writes, [],
|
|
"still read-only across both runs")
|
|
|
|
# 14 ----------------------------------------------------------------
|
|
def test_smoke_trap_refused(self):
|
|
self.gitea_state["token_mode"] = "empty"
|
|
self.gitea_state["releases"]["v0.2.3"] = {
|
|
"tag_name": "v0.2.3", "assets": [],
|
|
}
|
|
code, out, err = self.run_smoke("v0.2.3")
|
|
self.assertEqual(code, 1, "the trap is a loud remote failure")
|
|
self.assertIn("FAILED [2/4]", out)
|
|
self.assertIn("anonymous-200 trap refused", out)
|
|
self.assertIn("PASS [1/4]", out, "check 1 ran (order 1 -> 4)")
|
|
self.assertNotIn("PASS [3/4]", out, "checks after the trap never run")
|
|
self.assertNotIn("PASS [4/4]", out)
|
|
self.assertEqual(self._release_gets(), [],
|
|
"the release GET (step 3) is skipped")
|
|
self.assertEqual(self.hub_requests, [],
|
|
"zero hub requests: the index resolve (step 4) "
|
|
"is skipped")
|
|
|
|
# 15 ----------------------------------------------------------------
|
|
def test_smoke_missing_token_local(self):
|
|
empty_root = os.path.join(self.temp.name, "emptyroot")
|
|
os.makedirs(empty_root)
|
|
code, out, err = self.run_smoke("v0.2.3", env_root=empty_root)
|
|
self.assertEqual(code, 2, "local precondition -> exit 2")
|
|
self.assertIn("ERROR", err)
|
|
self.assertIn("GITEA_TOKEN", err)
|
|
self.assertEqual(self.gitea_state["requests"], [],
|
|
"zero gitea requests — the precondition fires "
|
|
"before anything remote")
|
|
self.assertEqual(self.hub_requests, [], "zero hub requests")
|
|
self.assertNotIn("IMAGE_PENDING", out, "no marker for local errors")
|
|
self.assertNotIn("PASS", out, "no smoke check ever ran")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main() |