import base64 import copy import hashlib import io import json import os import socket import subprocess import sys import tempfile import threading import unittest from contextlib import redirect_stderr, redirect_stdout from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path from urllib.parse import urlsplit from unittest.mock import patch import yaml ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "scripts")) import deploy import build import utils from utils import ( ConfigError, normalize_site_config, parse_site_yaml, validate_artifact_inputs, ) def fixture(name): return yaml.safe_load((ROOT / "tests" / "fixtures" / name).read_text()) class IPv6GitHTTPServer(ThreadingHTTPServer): address_family = socket.AF_INET6 class AuthenticatedGitHandler(BaseHTTPRequestHandler): project_root = None expected_authorization = None def log_message(self, _format, *_args): pass def do_GET(self): self._git_backend() def do_POST(self): self._git_backend() def _git_backend(self): if self.headers.get("Authorization") != self.expected_authorization: self.send_response(401) self.send_header("WWW-Authenticate", 'Basic realm="test"') self.end_headers() return parsed = urlsplit(self.path) length = int(self.headers.get("Content-Length", "0")) request_body = self.rfile.read(length) if length else b"" backend_env = os.environ.copy() backend_env.update({ "GIT_PROJECT_ROOT": str(self.project_root), "GIT_HTTP_EXPORT_ALL": "1", "PATH_INFO": parsed.path, "QUERY_STRING": parsed.query, "REQUEST_METHOD": self.command, "CONTENT_TYPE": self.headers.get("Content-Type", ""), "CONTENT_LENGTH": str(length), "REMOTE_USER": "ci-bot", "REMOTE_ADDR": "::1", "GATEWAY_INTERFACE": "CGI/1.1", "SERVER_PROTOCOL": "HTTP/1.1", }) result = subprocess.run( ["git", "http-backend"], input=request_body, env=backend_env, capture_output=True, check=True, ) raw_headers, response_body = result.stdout.split(b"\r\n\r\n", 1) headers, status = [], 200 for line in raw_headers.decode().split("\r\n"): name, value = line.split(":", 1) if name.lower() == "status": status = int(value.strip().split(" ", 1)[0]) else: headers.append((name, value.strip())) self.send_response(status) for name, value in headers: self.send_header(name, value) self.end_headers() self.wfile.write(response_body) class ConfigContractTests(unittest.TestCase): def setUp(self): self.raw = fixture("split-site.yaml") def assert_invalid(self, mutate, message): raw = copy.deepcopy(self.raw) mutate(raw) with self.assertRaisesRegex(ConfigError, message): normalize_site_config(raw, "baseline.fritzlab.net") def test_legacy_normalizes_to_explicit_compatibility_surface(self): cfg = normalize_site_config(fixture("legacy-site.yaml"), "example.fritzlab.net") self.assertEqual("single-surface-v1", cfg["compatibility"]) self.assertEqual("build/html", cfg["artifacts"][0]["build_dir"]) self.assertEqual("example.fritzlab.net", cfg["artifacts"][0]["bucket"]) self.assertEqual("/", cfg["routes"][0]["path"]) self.assertEqual("legacy", cfg["routes"][0]["access"]) def test_routes_are_sorted_longest_prefix_first(self): cfg = normalize_site_config(self.raw, "baseline.fritzlab.net") self.assertEqual(["/dist", "/"], [route["path"] for route in cfg["routes"]]) credentials = {artifact["name"]: artifact["credentials"] for artifact in cfg["artifacts"]} self.assertNotEqual(credentials["distributions"], credentials["portal"]) distributions = next( artifact for artifact in cfg["artifacts"] if artifact["name"] == "distributions" ) self.assertEqual(["*"], distributions["cors_origins"]) def test_cors_origins_are_https_origins_or_wildcard(self): self.assert_invalid( lambda raw: raw["artifacts"][1].__setitem__( "cors_origins", ["http://consumer.example"] ), "must contain '\\*' or canonical HTTPS origins", ) for origin in ( "https://consumer example", "https://Consumer.example", "https://consumer.example:443", "https://consumer.example/", "https://consumer.example:invalid", "https://[fe80::1%eth0]", "https://[fe80::1%25eth0]", ): with self.subTest(origin=origin): self.assert_invalid( lambda raw, origin=origin: raw["artifacts"][1].__setitem__( "cors_origins", [origin] ), "must contain '\\*' or canonical HTTPS origins", ) self.assert_invalid( lambda raw: raw["artifacts"][1].__setitem__( "cors_origins", ["https://consumer.example", "https://consumer.example:443"] ), "must not contain duplicate canonical origins", ) self.assert_invalid( lambda raw: raw["artifacts"][1].__setitem__( "cors_origins", ["*", "https://consumer.example"] ), "wildcard must be the only origin", ) raw = copy.deepcopy(self.raw) raw["artifacts"][1]["cors_origins"] = ["https://[2602:817:3000::1]:8443"] cfg = normalize_site_config(raw, "baseline.fritzlab.net") self.assertEqual( ["https://[2602:817:3000::1]:8443"], next(a for a in cfg["artifacts"] if a["name"] == "distributions")["cors_origins"], ) def test_declared_cors_origins_must_be_a_list(self): for value in (None, False, 0, "", {}): with self.subTest(value=value): self.assert_invalid( lambda raw, value=value: raw["artifacts"][1].__setitem__( "cors_origins", value ), "cors_origins must be a list", ) def test_protected_route_rejects_wildcard_cors(self): self.assert_invalid( lambda raw: raw["artifacts"][0].__setitem__("cors_origins", ["*"]), "protected route portal cannot allow wildcard CORS", ) def test_equivalent_route_paths_are_ambiguous(self): self.assert_invalid(lambda raw: raw["routes"].append({ "name": "duplicate", "path": "/dist/", "artifact": "portal", "access": {"mode": "protected", "middleware": "authentik-forwardauth"}, }), "ambiguous") def test_catch_all_is_required(self): self.assert_invalid(lambda raw: raw["routes"].__setitem__(0, { **raw["routes"][0], "path": "/portal" }), "catch-all") def test_public_catch_all_is_rejected_when_any_route_is_protected(self): def mutate(raw): raw["routes"][0]["access"] = {"mode": "public"} raw["routes"][1]["access"] = { "mode": "protected", "middleware": "authentik-forwardauth" } raw["artifacts"][1]["cors_origins"] = [] raw["artifacts"][0]["cache"]["rules"][0]["cache_control"] = ( "public, max-age=0, must-revalidate" ) for rule in raw["artifacts"][1]["cache"]["rules"]: rule["cache_control"] = "private, no-store" self.assert_invalid(mutate, "public '/' catch-all") def test_protected_route_requires_middleware(self): self.assert_invalid( lambda raw: raw["routes"][0].__setitem__("access", {"mode": "protected"}), "middleware is required", ) def test_bucket_cannot_cross_access_boundary(self): self.assert_invalid( lambda raw: raw["artifacts"][0]["publish"].__setitem__("bucket", "baseline-dist"), "protected and public", ) def test_publication_credentials_cannot_be_reused(self): def mutate(raw): raw["artifacts"][0]["publish"]["credentials"] = copy.deepcopy( raw["artifacts"][1]["publish"]["credentials"] ) self.assert_invalid(mutate, "publication credential DIST_S3_ACCESS_KEY is reused") def test_cache_directive_contradiction_is_rejected(self): self.assert_invalid( lambda raw: raw["artifacts"][1]["cache"]["rules"][1].__setitem__( "cache_control", "public, max-age=31536000, immutable, must-revalidate" ), "immutable requires", ) def test_cache_policy_cannot_nest_below_immutable_path(self): def mutate(raw): raw["artifacts"][1]["cache"]["rules"].append({ "path": "releases/candidates", "cache_control": "public, max-age=0, must-revalidate", }) self.assert_invalid(mutate, "cannot nest another policy under immutable /releases") def test_public_cache_is_rejected_on_protected_route(self): self.assert_invalid( lambda raw: raw["artifacts"][0]["cache"]["rules"][0].__setitem__( "cache_control", "public, max-age=0, must-revalidate" ), "protected route portal", ) def test_legacy_and_split_fields_cannot_mix(self): self.assert_invalid(lambda raw: raw.__setitem__("type", "static"), "cannot be mixed") def test_unknown_split_field_is_rejected(self): self.assert_invalid( lambda raw: raw["artifacts"][0].__setitem__("storage_bucket", "typo"), "unknown fields: storage_bucket", ) def test_backend_authority_and_endpoint_are_derived(self): for field, value in ( ("endpoint", "https://attacker.example"), ("website_authority", "internal-api.default.svc.k8s.sjc001.fritzlab.net"), ): with self.subTest(field=field): self.assert_invalid( lambda raw, field=field, value=value: raw["artifacts"][0]["publish"].__setitem__(field, value), f"unknown fields: {field}", ) def test_publication_credentials_use_dedicated_matched_names(self): self.assert_invalid( lambda raw: raw["artifacts"][0]["publish"]["credentials"].__setitem__( "access_key_env", "CI_BOT_TOKEN" ), "must be a matched _S3_ACCESS_KEY", ) def test_resolved_build_inputs_must_be_pairwise_disjoint(self): raw = copy.deepcopy(self.raw) raw["artifacts"][1]["content_dir"] = "" cfg = normalize_site_config(raw, "baseline.fritzlab.net") with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) (root / "portal" / "build").mkdir(parents=True) with self.assertRaisesRegex(ConfigError, "build inputs overlap after resolution"): validate_artifact_inputs(root, cfg) def test_descendant_symlink_cannot_cross_artifact_boundary(self): cfg = normalize_site_config(self.raw, "baseline.fritzlab.net") with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) (root / "dist").mkdir() (root / "portal" / "build").mkdir(parents=True) (root / "portal" / "build" / "private.txt").write_text("private") (root / "dist" / "portal-link").symlink_to(root / "portal" / "build") with self.assertRaisesRegex(ConfigError, "build input contains symlink"): validate_artifact_inputs(root, cfg) def test_artifact_root_symlink_is_rejected_before_resolution(self): cfg = normalize_site_config(self.raw, "baseline.fritzlab.net") with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) (root / "dist-real").mkdir() (root / "dist").symlink_to(root / "dist-real") (root / "portal" / "build").mkdir(parents=True) with self.assertRaisesRegex(ConfigError, "content_dir contains symlink component"): validate_artifact_inputs(root, cfg) class GenerationTests(unittest.TestCase): def render(self, raw): cfg = normalize_site_config(raw, "baseline.fritzlab.net") tmp = tempfile.TemporaryDirectory() root = Path(tmp.name) app_dir = root / "app" manifests = app_dir / "manifests" app_dir.mkdir(parents=True) deploy.render_site_manifests( "baseline.fritzlab.net", ROOT, app_dir, manifests, cfg ) files = {path.relative_to(app_dir).as_posix(): path.read_text() for path in sorted(app_dir.rglob("*.yaml"))} return tmp, app_dir, files def test_split_fixture_generates_per_route_resources_and_one_certificate(self): tmp, _, files = self.render(fixture("split-site.yaml")) self.addCleanup(tmp.cleanup) self.assertEqual({ "app.yaml", "manifests/certificate.yaml", "manifests/ingress-distributions.yaml", "manifests/ingress-portal.yaml", "manifests/kustomization.yaml", "manifests/service-distributions.yaml", "manifests/service-portal.yaml", "site-publish-history.yaml", }, set(files)) for content in files.values(): self.assertIsNotNone(yaml.safe_load(content)) self.assertIn("path: /dist", files["manifests/ingress-distributions.yaml"]) self.assertNotIn("authentik-forwardauth", files["manifests/ingress-distributions.yaml"]) self.assertIn("authentik-forwardauth@file", files["manifests/ingress-portal.yaml"]) self.assertIn('service.passhostheader: "false"', files["manifests/service-portal.yaml"]) self.assertNotIn("passhostheader", files["manifests/ingress-portal.yaml"]) self.assertIn("baseline-dist.web.sjc001.fritzlab.net", files["manifests/service-distributions.yaml"]) def test_yaml_ambiguous_artifact_name_stays_a_string_annotation(self): raw = fixture("split-site.yaml") raw["artifacts"][0]["name"] = "yes" raw["routes"][0]["artifact"] = "yes" tmp, _, files = self.render(raw) self.addCleanup(tmp.cleanup) ingress = yaml.safe_load(files["manifests/ingress-portal.yaml"]) self.assertEqual( "yes", ingress["metadata"]["annotations"]["site-publish.fritzlab.net/artifact"], ) def test_generation_is_deterministic_when_input_lists_are_reversed(self): raw = fixture("split-site.yaml") first_tmp, _, first = self.render(raw) self.addCleanup(first_tmp.cleanup) raw["artifacts"].reverse() raw["routes"].reverse() second_tmp, _, second = self.render(raw) self.addCleanup(second_tmp.cleanup) self.assertEqual(first, second) def test_stale_route_manifests_are_removed(self): raw = fixture("split-site.yaml") cfg = normalize_site_config(raw, "baseline.fritzlab.net") with tempfile.TemporaryDirectory() as tmp: app_dir = Path(tmp) / "app" manifests = app_dir / "manifests" manifests.mkdir(parents=True) stale = manifests / "ingress-removed.yaml" stale.write_text("stale\n") deploy.render_site_manifests("baseline.fritzlab.net", ROOT, app_dir, manifests, cfg) self.assertFalse(stale.exists()) def test_legacy_names_and_garage_s3_target_are_preserved(self): tmp, app_dir, files = self.render(fixture("legacy-site.yaml")) self.addCleanup(tmp.cleanup) self.assertIn("manifests/service.yaml", files) self.assertIn("manifests/ingress.yaml", files) self.assertIn("garage-s3.storage.svc.k8s.sjc001.fritzlab.net", files["manifests/service.yaml"]) self.assertNotIn("passhostheader", files["manifests/ingress.yaml"]) self.assertNotIn("site-publish.fritzlab.net", files["manifests/ingress.yaml"]) self.assertEqual({}, deploy.previous_route_contracts(app_dir)) class BuildTests(unittest.TestCase): def test_artifacts_build_independently_without_clobbering_siblings(self): raw = fixture("split-site.yaml") for artifact in raw["artifacts"]: artifact["tidy"] = False cfg = normalize_site_config(raw, "baseline.fritzlab.net") with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) (root / "dist").mkdir() (root / "dist" / "bundle.js").write_text("bundle") (root / "portal" / "build").mkdir(parents=True) (root / "portal" / "build" / "index.html").write_text("portal") for artifact in cfg["artifacts"]: build.build_artifact(root, artifact) self.assertEqual( "bundle", (root / ".site-publish/distributions/html/bundle.js").read_text() ) self.assertEqual( "portal", (root / ".site-publish/portal/html/index.html").read_text() ) class ArtifactSelectionTests(unittest.TestCase): """The `artifacts` input scopes publication, never the route contract.""" def parse(self, selection): with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) (root / "site.yaml").write_text( (ROOT / "tests" / "fixtures" / "split-site.yaml").read_text() ) environment = {"SITE_REPO": "fritzlab/baseline"} if selection is not None: environment["SITE_ARTIFACTS"] = selection with patch.dict(os.environ, environment, clear=False), \ redirect_stdout(io.StringIO()): if selection is None: os.environ.pop("SITE_ARTIFACTS", None) return parse_site_yaml(root) def test_absent_selection_publishes_every_declared_artifact(self): self.assertEqual(["distributions", "portal"], self.parse(None)["selected"]) self.assertEqual(["distributions", "portal"], self.parse(" ")["selected"]) def test_selection_accepts_comma_and_whitespace_separated_names(self): self.assertEqual(["portal"], self.parse("portal")["selected"]) self.assertEqual( ["distributions", "portal"], self.parse("portal, distributions")["selected"] ) self.assertEqual( ["distributions", "portal"], self.parse("portal distributions")["selected"] ) def test_undeclared_selection_fails_before_any_publication(self): stderr = io.StringIO() with redirect_stderr(stderr), self.assertRaises(SystemExit): self.parse("catalogue") self.assertIn("undeclared artifact", stderr.getvalue()) self.assertIn("catalogue", stderr.getvalue()) def test_build_skips_unselected_artifacts(self): raw = fixture("split-site.yaml") for artifact in raw["artifacts"]: artifact["tidy"] = False cfg = normalize_site_config(raw, "baseline.fritzlab.net") cfg["selected"] = ["portal"] with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) (root / "portal" / "build").mkdir(parents=True) (root / "portal" / "build" / "index.html").write_text("portal") built = [] with patch.object(build, "build_artifact", side_effect=lambda _root, artifact: built.append(artifact["name"])), \ patch.dict(os.environ, {"SITE_DIR": str(root)}, clear=False), \ patch.object(build, "parse_site_yaml", return_value=cfg), \ redirect_stdout(io.StringIO()): build.cmd_build() self.assertEqual(["portal"], built) def deploy_selection(self, selection, environment): cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net") cfg["selected"] = selection tmp = tempfile.TemporaryDirectory() self.addCleanup(tmp.cleanup) root = Path(tmp.name) apps_dir = root / "apps" app_dir = apps_dir / "sjc001" / "websites" / "baseline.fritzlab.net" app_dir.mkdir(parents=True) # A scoped run refuses an artifact it has no published history for, so # the fixture starts from the whole publication this one narrows. (app_dir / "site-publish-history.yaml").write_text(yaml.safe_dump({ "version": 1, "buckets": { artifact["bucket"]: { "path": route["path"], "access": "public" if route["access"] == "legacy" else route["access"], "artifact": route["artifact"], "immutable_paths": deploy.immutable_key_prefixes(artifact, route), } for route in cfg["routes"] for artifact in [next(item for item in cfg["artifacts"] if item["name"] == route["artifact"])] }, }, sort_keys=True)) for artifact in cfg["artifacts"]: if artifact["name"] not in set(selection): continue html = root / artifact["build_dir"] html.mkdir(parents=True) (html / "index.html").write_text(artifact["name"]) for rule in artifact["cache_rules"]: if rule["path"]: (html / rule["path"]).mkdir(parents=True, exist_ok=True) (html / rule["path"] / "keep.json").write_text("{}") synced, immutables, cors = [], [], [] with patch.dict(os.environ, environment, clear=False), \ patch.object(deploy, "clone_apps", return_value=apps_dir), \ patch.object(deploy, "commit_and_push"), \ patch.object(deploy, "publish_route_immutables", side_effect=lambda artifact, *_a, **_k: immutables.append(artifact["name"])), \ patch.object(deploy, "reconcile_artifact_cors", side_effect=lambda artifacts, *_a: cors.extend(item["name"] for item in artifacts)), \ patch.object(deploy, "s3_sync", side_effect=lambda artifact, *_a, **_k: synced.append(artifact["name"])), \ redirect_stdout(io.StringIO()): deploy.deploy_static("baseline.fritzlab.net", root, ROOT, "token", cfg) files = {path.relative_to(app_dir).as_posix(): path.read_text() for path in sorted(app_dir.rglob("*.yaml"))} return {"synced": synced, "immutables": immutables, "cors": cors, "files": files} def test_selected_publication_leaves_the_whole_route_contract_rendered(self): # Only the portal credentials exist: a scoped run must not demand the # secrets of an artifact it is not publishing. environment = {"PORTAL_S3_ACCESS_KEY": "portal-key", "PORTAL_S3_SECRET_KEY": "portal-secret"} for name in ("DIST_S3_ACCESS_KEY", "DIST_S3_SECRET_KEY"): os.environ.pop(name, None) result = self.deploy_selection(["portal"], environment) self.assertEqual(["portal"], result["synced"]) self.assertEqual(["portal"], result["immutables"]) self.assertEqual(["portal"], result["cors"]) self.assertIn("manifests/ingress-distributions.yaml", result["files"]) self.assertIn("manifests/service-distributions.yaml", result["files"]) history = yaml.safe_load(result["files"]["site-publish-history.yaml"]) self.assertEqual({"baseline-dist", "baseline-portal"}, set(history["buckets"])) self.assertEqual( ["dist/releases"], history["buckets"]["baseline-dist"]["immutable_paths"] ) def test_scoped_run_refuses_to_record_a_contract_it_did_not_publish(self): """The reviewer's scenario: distributions goes protected in site.yaml while a catalogue-only run publishes. Recording that access would make the bucket unreturnable through validate_route_migrations.""" cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net") published = { "baseline-portal": { "path": "/", "access": "protected", "artifact": "portal", "immutable_paths": [], }, "baseline-dist": { "path": "/dist", "access": "public", "artifact": "distributions", "immutable_paths": ["dist/releases"], }, } # A whole publish agrees with what is recorded, scoped or not. deploy.validate_scoped_history(cfg, published) cfg["selected"] = ["portal"] deploy.validate_scoped_history(cfg, published) moved = copy.deepcopy(cfg) route = next(item for item in moved["routes"] if item["artifact"] == "distributions") route["access"] = "protected" route["access_middleware"] = "authentik-forwardauth" with self.assertRaises(RuntimeError) as refused: deploy.validate_scoped_history(moved, published) self.assertIn("may not change unselected artifact distributions", str(refused.exception)) # The same change published in the same run is allowed to proceed. moved["selected"] = ["distributions", "portal"] deploy.validate_scoped_history(moved, published) def test_scoped_run_refuses_an_artifact_with_no_published_history(self): cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net") cfg["selected"] = ["portal"] with self.assertRaises(RuntimeError) as refused: deploy.validate_scoped_history(cfg, {}) self.assertIn("cannot introduce artifact distributions", str(refused.exception)) def test_scoped_deploy_runs_the_history_guard_before_touching_a_bucket(self): cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net") cfg["selected"] = ["portal"] tmp = tempfile.TemporaryDirectory() self.addCleanup(tmp.cleanup) root = Path(tmp.name) apps_dir = root / "apps" app_dir = apps_dir / "sjc001" / "websites" / "baseline.fritzlab.net" app_dir.mkdir(parents=True) html = root / next( item for item in cfg["artifacts"] if item["name"] == "portal" )["build_dir"] html.mkdir(parents=True) (html / "index.html").write_text("portal") with patch.dict(os.environ, { "PORTAL_S3_ACCESS_KEY": "portal-key", "PORTAL_S3_SECRET_KEY": "portal-secret", }, clear=False), \ patch.object(deploy, "clone_apps", return_value=apps_dir), \ patch.object(deploy, "commit_and_push"), \ patch.object(deploy, "publish_route_immutables") as immutables, \ patch.object(deploy, "reconcile_artifact_cors") as cors, \ patch.object(deploy, "s3_sync") as sync, \ redirect_stdout(io.StringIO()), \ self.assertRaises(RuntimeError): deploy.deploy_static("baseline.fritzlab.net", root, ROOT, "token", cfg) immutables.assert_not_called() cors.assert_not_called() sync.assert_not_called() def test_disabled_site_refuses_a_partial_selection(self): cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net") cfg["enabled"] = False cfg["selected"] = ["portal"] with patch.object(deploy, "parse_site_yaml", return_value=cfg), \ patch.dict(os.environ, { "SITE_REPO": "fritzlab/baseline.fritzlab.net", "SITE_DIR": ".", "ACTION_DIR": str(ROOT), "CI_BOT_TOKEN": "token", }, clear=False), \ patch.object(deploy, "decommission") as decommission, \ redirect_stderr(io.StringIO()), redirect_stdout(io.StringIO()), \ self.assertRaises(SystemExit): deploy.cmd_deploy() decommission.assert_not_called() class PublishingTests(unittest.TestCase): def test_cors_policy_is_reconciled_as_read_only_browser_access(self): captured = {} def capture(command, **_kwargs): config_path = command[command.index("--cors-configuration") + 1] captured["command"] = command captured["config"] = json.loads(Path(config_path.removeprefix("file://")).read_text()) with patch.object(deploy, "run", side_effect=capture): deploy.configure_cors( "baseline-dist", ["*"], "http://garage-s3.storage.svc:3900", {} ) self.assertIn("put-bucket-cors", captured["command"]) self.assertEqual(["GET", "HEAD"], captured["config"]["CORSRules"][0]["AllowedMethods"]) self.assertEqual(["*"], captured["config"]["CORSRules"][0]["AllowedOrigins"]) def test_empty_cors_policy_removes_stale_bucket_cors(self): with patch.object(deploy, "run") as request: deploy.configure_cors( "baseline-catalogue", [], "http://garage-s3.storage.svc:3900", {} ) self.assertIn("delete-bucket-cors", request.call_args.args[0]) def test_cors_reconciliation_restores_prior_policies_on_failure(self): artifacts = [ { "bucket": "first", "cors_origins": ["https://new.example"], "s3_endpoint": "http://garage-s3.storage.svc:3900", "credentials": {}, }, { "bucket": "second", "cors_origins": [], "s3_endpoint": "http://garage-s3.storage.svc:3900", "credentials": {}, }, ] prior = [ {"CORSRules": [{"AllowedOrigins": ["https://old.example"]}]}, None, ] writes = [] def write(bucket, config, *_args): writes.append((bucket, config)) if bucket == "second" and len(writes) == 2: raise RuntimeError("write failed") with patch.object(deploy, "publication_aws_env", return_value={}), patch.object( deploy, "get_cors_configuration", side_effect=prior, ) as read, patch.object(deploy, "set_cors_configuration", side_effect=write), \ self.assertRaisesRegex(RuntimeError, "write failed"): deploy.reconcile_artifact_cors(artifacts) self.assertEqual(2, read.call_count) self.assertEqual( [ ("first", {"CORSRules": [{ "AllowedOrigins": ["https://new.example"], "AllowedMethods": ["GET", "HEAD"], "AllowedHeaders": ["*"], "ExposeHeaders": ["ETag"], "MaxAgeSeconds": 3600, }]}), ("second", None), ("second", None), ("first", prior[0]), ], writes, ) def test_apps_clone_never_places_token_in_argv_or_log(self): calls = [] secret = "clone-secret-must-not-appear" with patch.dict(os.environ, {"CI_BOT_USER": "ci-bot"}, clear=False), \ patch.object(utils, "run", side_effect=lambda command, **kwargs: calls.append((command, kwargs))), \ patch.object(utils.shutil, "rmtree"), redirect_stdout(io.StringIO()) as output: utils.clone_apps(secret) self.assertNotIn(secret, output.getvalue()) for command, kwargs in calls: self.assertNotIn(secret, " ".join(command)) self.assertNotIn(secret, kwargs.get("display", "")) def test_askpass_authenticates_clone_and_push_round_trip(self): token = "round-trip-token" expected = "Basic " + base64.b64encode(f"ci-bot:{token}".encode()).decode() with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) bare = root / "repo.git" seed = root / "seed" checkout = root / "checkout" subprocess.run(["git", "init", "--bare", "--initial-branch=main", str(bare)], check=True, stdout=subprocess.DEVNULL) subprocess.run(["git", "-C", str(bare), "config", "http.receivepack", "true"], check=True) subprocess.run(["git", "init", "--initial-branch=main", str(seed)], check=True, stdout=subprocess.DEVNULL) subprocess.run(["git", "-C", str(seed), "config", "user.name", "Test"], check=True) subprocess.run(["git", "-C", str(seed), "config", "user.email", "test@example.invalid"], check=True) (seed / "README.md").write_text("seed\n") subprocess.run(["git", "-C", str(seed), "add", "README.md"], check=True) subprocess.run(["git", "-C", str(seed), "commit", "-m", "seed"], check=True, stdout=subprocess.DEVNULL) subprocess.run(["git", "-C", str(seed), "push", str(bare), "main"], check=True, stdout=subprocess.DEVNULL) handler = type("GitHandler", (AuthenticatedGitHandler,), { "project_root": root, "expected_authorization": expected, }) server = IPv6GitHTTPServer(("::1", 0), handler) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() try: url = f"http://ci-bot@[::1]:{server.server_port}/repo.git" with patch.dict(os.environ, {"NO_PROXY": "::1,[::1]", "no_proxy": "::1,[::1]"}, clear=False): subprocess.run( ["git", "clone", url, str(checkout)], env=utils.git_auth_env(token), check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, ) subprocess.run(["git", "-C", str(checkout), "config", "user.name", "Test"], check=True) subprocess.run(["git", "-C", str(checkout), "config", "user.email", "test@example.invalid"], check=True) (checkout / "roundtrip.txt").write_text("authenticated\n") utils.commit_and_push(checkout, "authenticated round trip", token) finally: server.shutdown() server.server_close() thread.join(timeout=5) result = subprocess.run( ["git", "--git-dir", str(bare), "show", "main:roundtrip.txt"], check=True, text=True, capture_output=True, ) self.assertEqual("authenticated\n", result.stdout) def test_cache_headers_credentials_and_route_prefix_are_separate(self): cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net") artifact = next(item for item in cfg["artifacts"] if item["name"] == "distributions") route = next(item for item in cfg["routes"] if item["artifact"] == "distributions") with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) html = root / artifact["build_dir"] (html / "releases").mkdir(parents=True) (html / "channels").mkdir() (html / "releases" / "1.0.js").write_text("release") (html / "channels" / "stable.json").write_text("channel") commands, events = [], [] def capture(command, **kwargs): commands.append((command, kwargs["env"])) events.append("mutable") def publish_immutable(*_args): events.append("immutable") secret = "secret-must-not-appear" output = io.StringIO() with patch.dict(os.environ, { "DIST_S3_ACCESS_KEY": "dist-key", "DIST_S3_SECRET_KEY": secret }, clear=False), patch.object(deploy, "run", side_effect=capture), \ patch.object(deploy, "publish_immutable_rule", side_effect=publish_immutable) as immutable_publish, \ redirect_stdout(output): deploy.publish_route_immutables(artifact, route, root) deploy.s3_sync(artifact, route, root) self.assertTrue(all(secret not in " ".join(command) for command, _ in commands)) self.assertEqual("immutable", events[0]) self.assertNotIn(secret, output.getvalue()) self.assertTrue(all(call_env["AWS_ACCESS_KEY_ID"] == "dist-key" for _, call_env in commands)) self.assertTrue(all("DIST_S3_SECRET_KEY" not in call_env for _, call_env in commands)) rendered = [" ".join(command) for command, _ in commands] self.assertIn("s3://baseline-dist/dist/", rendered[0]) self.assertIn("releases/*", rendered[0]) self.assertNotIn("--delete", rendered[0]) self.assertIn("--delete", rendered[1]) self.assertTrue(all("s3://baseline-dist/dist/" in command for command in rendered[1:])) self.assertTrue(any("channels/" in command and "public, max-age=0, must-revalidate" in command for command in rendered)) immutable_publish.assert_called_once() self.assertEqual( "public, max-age=31536000, immutable", immutable_publish.call_args.args[2]["cache_control"], ) def test_root_move_preserves_only_actual_retired_immutable_prefix(self): cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net") artifact = next(item for item in cfg["artifacts"] if item["name"] == "distributions") route = {**next(item for item in cfg["routes"] if item["artifact"] == "distributions"), "path": "/"} with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) html = root / artifact["build_dir"] (html / "releases").mkdir(parents=True) (html / "channels").mkdir() (html / "docs" / "releases").mkdir(parents=True) (html / "channels" / "stable.json").write_text("channel") (html / "docs" / "releases" / "index.html").write_text("mutable") commands = [] with patch.dict(os.environ, { "DIST_S3_ACCESS_KEY": "dist-key", "DIST_S3_SECRET_KEY": "dist-secret" }, clear=False), patch.object( deploy, "run", side_effect=lambda command, **_: commands.append(command) ): deploy.s3_sync(artifact, route, root, previous_contract={ "path": "/foo", "access": "public", "artifact": "distributions", "immutable_paths": ["foo/releases"], }) rendered = [" ".join(command) for command in commands] self.assertTrue(all("foo/releases/*" in command for command in rendered[:2])) self.assertTrue(all("*/releases/*" not in command for command in rendered)) def test_protected_bucket_cannot_become_public_across_deployments(self): cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net") previous = { "baseline-dist": { "path": "/dist", "access": "protected", "artifact": "old-name", "immutable_paths": ["dist/releases"], } } with self.assertRaisesRegex(RuntimeError, "cannot become public while reusing protected"): deploy.validate_route_migrations(cfg, previous) renamed = copy.deepcopy(cfg) artifact = next(item for item in renamed["artifacts"] if item["name"] == "distributions") artifact["name"] = "downloads" route = next(item for item in renamed["routes"] if item["artifact"] == "distributions") route["artifact"] = "downloads" with self.assertRaisesRegex(RuntimeError, "cannot become public while reusing protected"): deploy.validate_route_migrations(renamed, previous) previous = {"retired-protected-bucket": previous["baseline-dist"]} deploy.validate_route_migrations(cfg, previous) def test_protected_split_bucket_cannot_become_legacy_public(self): cfg = normalize_site_config(fixture("legacy-site.yaml"), "baseline.fritzlab.net") previous = { "baseline.fritzlab.net": { "path": "/portal", "access": "protected", "artifact": "portal", "immutable_paths": [], } } with self.assertRaisesRegex(RuntimeError, "cannot become public while reusing protected"): deploy.validate_route_migrations(cfg, previous) def test_removed_immutable_rule_preserves_prior_keys(self): cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net") artifact = next(item for item in cfg["artifacts"] if item["name"] == "distributions") artifact["cache_rules"] = [ rule for rule in artifact["cache_rules"] if rule["path"] != "releases" ] route = next(item for item in cfg["routes"] if item["artifact"] == "distributions") with tempfile.TemporaryDirectory() as tmp: html = Path(tmp) filters = deploy.retired_immutable_filters(artifact, route, html, { "path": "/dist", "access": "public", "artifact": "distributions", "immutable_paths": ["dist/releases"], }) self.assertEqual(["--exclude", "releases/*"], filters) (html / "releases").mkdir() (html / "releases" / "replacement.js").write_text("mutable") with self.assertRaisesRegex(RuntimeError, "collides with retired immutable"): deploy.retired_immutable_filters(artifact, route, html, { "path": "/dist", "access": "public", "artifact": "distributions", "immutable_paths": ["dist/releases"], }) def test_removed_immutable_rule_remains_in_next_manifest_history(self): cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net") artifact = next(item for item in cfg["artifacts"] if item["name"] == "distributions") artifact["cache_rules"] = [ rule for rule in artifact["cache_rules"] if rule["path"] != "releases" ] previous = { "baseline-dist": { "path": "/dist", "access": "public", "artifact": "distributions", "immutable_paths": ["dist/releases"], } } with tempfile.TemporaryDirectory() as tmp: app_dir = Path(tmp) / "app" manifests = app_dir / "manifests" deploy.render_site_manifests( "baseline.fritzlab.net", ROOT, app_dir, manifests, cfg, previous, ) first = deploy.previous_route_contracts(app_dir) self.assertEqual(["dist/releases"], first["baseline-dist"]["immutable_paths"]) deploy.render_site_manifests( "baseline.fritzlab.net", ROOT, app_dir, manifests, cfg, first, ) second = deploy.previous_route_contracts(app_dir) self.assertEqual(first, second) def test_protected_split_bucket_survives_absence_and_blocks_legacy(self): raw = fixture("split-site.yaml") raw["artifacts"] = [ item for item in raw["artifacts"] if item["name"] == "distributions" ] raw["routes"] = [ item for item in raw["routes"] if item["artifact"] == "distributions" ] raw["routes"][0]["path"] = "/" cfg = normalize_site_config(raw, "baseline.fritzlab.net") previous = { "baseline.fritzlab.net": { "path": "/", "access": "protected", "artifact": "portal", "immutable_paths": [], } } with tempfile.TemporaryDirectory() as tmp: app_dir = Path(tmp) / "app" deploy.render_site_manifests( "baseline.fritzlab.net", ROOT, app_dir, app_dir / "manifests", cfg, previous, ) retained = deploy.previous_route_contracts(app_dir) self.assertEqual( previous["baseline.fritzlab.net"], retained["baseline.fritzlab.net"], ) legacy = normalize_site_config( fixture("legacy-site.yaml"), "baseline.fritzlab.net", ) with self.assertRaisesRegex(RuntimeError, "cannot become public while reusing protected"): deploy.validate_route_migrations(legacy, retained) def test_malformed_persistent_history_fails_closed(self): with tempfile.TemporaryDirectory() as tmp: app_dir = Path(tmp) (app_dir / "site-publish-history.yaml").write_text( "version: 1\nbuckets:\n bucket:\n path: /\n" " access: public\n artifact: site\n" " immutable_paths: [../releases]\n" ) with self.assertRaisesRegex(RuntimeError, "invalid site-publish route history"): deploy.previous_route_contracts(app_dir) def test_retired_immutable_history_survives_a_route_move(self): cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net") artifact = next(item for item in cfg["artifacts"] if item["name"] == "distributions") artifact["cache_rules"] = [ rule for rule in artifact["cache_rules"] if rule["path"] != "releases" ] route = next(item for item in cfg["routes"] if item["artifact"] == "distributions") route["path"] = "/" previous = { "baseline-dist": { "path": "/foo", "access": "public", "artifact": "distributions", "immutable_paths": ["foo/releases"], } } with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) app_dir = root / "app" manifests = app_dir / "manifests" app_dir.mkdir() deploy.render_site_manifests( "baseline.fritzlab.net", ROOT, app_dir, manifests, cfg, previous, ) following = deploy.previous_route_contracts(app_dir) self.assertEqual(["foo/releases"], following["baseline-dist"]["immutable_paths"]) html = root / "html" html.mkdir() first_filters = deploy.retired_immutable_filters( artifact, route, html, previous["baseline-dist"], ) following_filters = deploy.retired_immutable_filters( artifact, route, html, following["baseline-dist"], ) self.assertEqual(["--exclude", "foo/releases/*"], first_filters) self.assertEqual(first_filters, following_filters) def test_later_route_immutable_failure_stops_all_mutable_publication(self): cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net") with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) with patch.object(deploy, "validate_publication_environment"), \ patch.object(deploy, "validate_artifact_output"), \ patch.object(deploy, "clone_apps", return_value=root / "apps"), patch.object( deploy, "publish_route_immutables", side_effect=[None, RuntimeError("immutable failed")], ) as immutable_publish, patch.object( deploy, "reconcile_artifact_cors" ) as cors_reconcile, patch.object(deploy, "s3_sync") as mutable_sync, \ self.assertRaisesRegex( RuntimeError, "immutable failed" ): deploy.deploy_static("baseline", root, root, "token", cfg) self.assertEqual(2, immutable_publish.call_count) cors_reconcile.assert_not_called() mutable_sync.assert_not_called() def test_all_cors_policies_complete_before_mutable_publication(self): cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net") events = [] with tempfile.TemporaryDirectory() as tmp: root = Path(tmp) apps = root / "apps" apps.mkdir() with patch.object(deploy, "validate_publication_environment"), patch.object( deploy, "validate_artifact_output" ), patch.object( deploy, "publish_route_immutables", side_effect=lambda *_args: events.append("immutable") ), patch.object( deploy, "reconcile_artifact_cors", side_effect=lambda *_args: events.append("cors") ), patch.object( deploy, "s3_sync", side_effect=lambda *_args: events.append("mutable") ), patch.object(deploy, "clone_apps", return_value=apps), patch.object( deploy, "render_site_manifests" ), patch.object(deploy, "commit_and_push"): deploy.deploy_static("baseline", root, root, "token", cfg) self.assertEqual( ["immutable", "immutable", "cors", "mutable", "mutable"], events ) def test_absent_artifact_is_detected_before_publish(self): cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net") with tempfile.TemporaryDirectory() as tmp, redirect_stderr(io.StringIO()), \ self.assertRaises(SystemExit): deploy.validate_artifact_output(Path(tmp), cfg["artifacts"][0]) def test_absent_cache_prefix_is_detected(self): cfg = normalize_site_config(fixture("split-site.yaml"), "baseline.fritzlab.net") artifact = next(item for item in cfg["artifacts"] if item["name"] == "distributions") with tempfile.TemporaryDirectory() as tmp: html = Path(tmp) / artifact["build_dir"] html.mkdir(parents=True) (html / "index.html").write_text("content") with redirect_stderr(io.StringIO()), self.assertRaises(SystemExit): deploy.validate_artifact_output(Path(tmp), artifact) class ImmutablePublicationTests(unittest.TestCase): CACHE = "public, max-age=31536000, immutable" def result(self, returncode, stdout="", stderr=""): return subprocess.CompletedProcess([], returncode, stdout, stderr) def key_and_digests(self, source): content_type = "text/javascript" content_digest, publication_digest = deploy._immutable_digests( source, self.CACHE, content_type, ) return f"dist/releases/release-{publication_digest}.js", content_digest, publication_digest def test_new_key_uses_content_address_and_digest_metadata(self): with tempfile.TemporaryDirectory() as tmp: source = Path(tmp) / "release.js" source.write_text("fixed release") key, content_digest, publication_digest = self.key_and_digests(source) calls = [] def capture(args, _env): calls.append(args) return self.result(1, stderr="404 Not Found") if len(calls) == 1 else self.result(0, "{}") with patch.object(deploy, "_aws_capture", side_effect=capture): created = deploy.publish_immutable_file( "http://garage-s3.storage.svc:3900", "dist-bucket", key, source, self.CACHE, {}, ) self.assertTrue(created) put = calls[1] self.assertEqual("put-object", put[4]) self.assertNotIn("--if-none-match", put) self.assertEqual( f"sha256={content_digest},publication-sha256={publication_digest}", put[put.index("--metadata") + 1], ) def test_identical_retry_converges_without_put(self): with tempfile.TemporaryDirectory() as tmp: source = Path(tmp) / "release.js" source.write_text("fixed release") key, content_digest, publication_digest = self.key_and_digests(source) head = json.dumps({ "Metadata": {"sha256": content_digest, "publication-sha256": publication_digest}, "CacheControl": self.CACHE, "ContentType": "text/javascript", }) with patch.object(deploy, "_aws_capture", return_value=self.result(0, head)) as request: created = deploy.publish_immutable_file( "http://garage-s3.storage.svc:3900", "dist-bucket", key, source, self.CACHE, {}, ) self.assertFalse(created) self.assertEqual(1, request.call_count) def test_changed_immutable_key_is_refused(self): with tempfile.TemporaryDirectory() as tmp: source = Path(tmp) / "release.js" source.write_text("changed release") key, _, _ = self.key_and_digests(source) head = json.dumps({"Metadata": {"sha256": "different"}, "CacheControl": self.CACHE}) with patch.object(deploy, "_aws_capture", return_value=self.result(0, head)), \ self.assertRaisesRegex(RuntimeError, "immutable object differs"): deploy.publish_immutable_file( "http://garage-s3.storage.svc:3900", "dist-bucket", key, source, self.CACHE, {}, ) def test_immutable_key_without_publication_digest_is_refused_before_s3(self): with tempfile.TemporaryDirectory() as tmp: source = Path(tmp) / "release.js" source.write_text("fixed release") with patch.object(deploy, "_aws_capture") as request, \ self.assertRaisesRegex(RuntimeError, "must contain its one publication SHA-256"): deploy.publish_immutable_file( "http://garage-s3.storage.svc:3900", "dist-bucket", "dist/releases/release.js", source, self.CACHE, {}, ) request.assert_not_called() if __name__ == "__main__": unittest.main()