feat: add split-surface site publishing

2026-08-29 21:22:04 +00:00
7 changed files with 70 additions and 1245 deletions
@@ -52,7 +52,6 @@ artifacts:
  - name: distributions
    type: static
    content_dir: dist
    cors_origins: ['*']
    publish:
      bucket: baseline-dist
      credentials:
@@ -93,10 +92,8 @@ routes:
 ```
 The caller supplies each declared credential name as an environment variable
-on the action step. Names must be matched `<NAME>_S3_ACCESS_KEY` and
+on the action step. Credential values are passed to `aws` only through its
-`<NAME>_S3_SECRET_KEY` pairs; arbitrary environment variables cannot become
+environment and never appear in a logged command or process argument.
 publication credentials. Values pass to `aws` only through its environment and
 never appear in a logged command or process argument.
 ```yaml
      - uses: https://code.fritzlab.net/action/site-publish@v1
@@ -123,40 +120,6 @@ revalidation, or `no-store` plus a positive max-age) are rejected. Protected
 artifacts require `private` or `no-store` and cannot emit `public`.
 Metadata restamping transfers each artifact once even on a no-op publication;
 that is the cost of making policy changes effective on unchanged Garage objects.
 An immutable cache path is excluded from sync and deletion. Every object key in
 that path must contain exactly one full publication SHA-256, calculated over its
 cache policy, content type, and bytes. That content address makes concurrent
 writes identical even though Garage v2.2.0 has no conditional destination
 write. An identical retry converges; a changed object, missing digest metadata,
 wrong address, or nested policy under that immutable prefix fails publication.
 Every immutable target across every artifact is validated and published before
 any route's mutable objects change.
 Mutable default and override partitions receive their final cache policy before
 the matching prefix-scoped stale deletion, so publication never exposes a
 provisional cache policy or a pointer to a missing immutable target.
 Generated Ingress annotations and `site-publish-history.yaml` retain every
 seen bucket's access, prior route, and cumulative bucket-relative immutable key
 prefixes, including while an artifact is absent. Removed or renamed rules stay
 recorded. When a move places a retired prefix inside the new sync scope, its
 immutable subtrees are excluded; a current-file collision fails publication.
 The bucket-keyed history rejects a protected-to-public transition even when the
 artifact is renamed; publishing that artifact publicly requires a new bucket.
 Legacy single-surface is public for this downgrade check. Removing or renaming
 an immutable rule preserves its prior URLs; current mutable content at one of
 those paths is rejected instead of replacing it.
 Artifact input directories must be pairwise disjoint after filesystem
 resolution. Publication stops before build or upload if one contains another or
 escapes the repository. Symlinked roots, components, and descendants are also rejected, preventing
 protected input from entering a public artifact through dereference. Split
 storage endpoints are pinned to Garage, and each website
 authority is derived from its bucket; a site cannot expose an arbitrary backend.
 `cors_origins` is reconciled as a bucket policy on every split publication. Values are either `*`
 or HTTPS origins; browser access is limited to `GET` and `HEAD`. Omitting the field removes stale
 CORS from that bucket. Protected artifacts cannot allow wildcard CORS. All immutable objects and
 all bucket CORS policies complete before mutable channels change; if any policy write fails, the
 policies already attempted are restored to their pre-publication values.
 Each split route gets a bucket-specific `<bucket>.web.sjc001.fritzlab.net`
 ExternalName Service annotated to disable pass-host-header and a separate Ingress. Route
@@ -5,7 +5,7 @@ import subprocess
 import tempfile
 from pathlib import Path
-from utils import EXCLUDE_FILES, env, parse_site_yaml, run, validate_artifact_inputs
+from utils import EXCLUDE_FILES, env, parse_site_yaml, run
 def build_artifact(site_dir, artifact):
@@ -64,7 +64,5 @@ def cmd_build():
        print("Site disabled — skipping build")
        return
    validate_artifact_inputs(site_dir, cfg)
    for artifact in cfg["artifacts"]:
        build_artifact(site_dir, artifact)
@@ -1,20 +1,13 @@
 """Deploy phase — S3 sync, manifest rendering, alias reconcile."""
 import fnmatch
 import hashlib
 import json
 import mimetypes
 import os
 import re
 import shutil
 import subprocess
 import tempfile
 from pathlib import Path
 from urllib.error import HTTPError, URLError
 from urllib.request import Request, urlopen
 import yaml
 from utils import (
    NAMESPACE,
    clone_apps,
@@ -25,7 +18,6 @@ from utils import (
    parse_site_yaml,
    render_templates,
    run,
    validate_artifact_inputs,
 )
 GARAGE_ADMIN_ENDPOINT = os.environ.get(
@@ -55,105 +47,9 @@ def validate_publication_environment(cfg):
        die("GARAGE_ADMIN_TOKEN is required when aliases are declared")
-def _is_immutable(rule):
+def s3_sync(artifact, route, site_dir, credential_env_names=None):
-    return "immutable" in {
+    endpoint = artifact["s3_endpoint"]
-        part.strip().lower().split("=", 1)[0]
+    html_dir = site_dir / artifact["build_dir"]
        for part in rule["cache_control"].split(",")
    }
 def _aws_capture(args, aws_env):
    """Run a non-streaming AWS request without exposing environment credentials."""
    print(f"  $ {' '.join(str(part) for part in args)}")
    return subprocess.run(args, env=aws_env, text=True, capture_output=True, check=False)
 def _immutable_head(endpoint, bucket, key, aws_env):
    args = ["aws", "--endpoint-url", endpoint, "s3api", "head-object",
            "--bucket", bucket, "--key", key, "--output", "json"]
    result = _aws_capture(args, aws_env)
    if result.returncode == 0:
        return json.loads(result.stdout)
    error = f"{result.stdout}\n{result.stderr}"
    if any(marker in error for marker in ("404", "Not Found", "NoSuchKey")):
        return None
    raise RuntimeError(f"head-object failed for s3://{bucket}/{key}: {error.strip()}")
 def _immutable_digests(source, cache_control, content_type):
    with source.open("rb") as stream:
        content_digest = hashlib.file_digest(stream, "sha256").hexdigest()
    publication = hashlib.sha256()
    publication.update(cache_control.encode())
    publication.update(b"\0")
    publication.update(content_type.encode())
    publication.update(b"\0")
    with source.open("rb") as stream:
        for block in iter(lambda: stream.read(1024 * 1024), b""):
            publication.update(block)
    return content_digest, publication.hexdigest()
 def _same_immutable_object(info, content_digest, publication_digest, cache_control, content_type):
    metadata = {key.lower(): value for key, value in (info.get("Metadata") or {}).items()}
    return (
        metadata.get("sha256") == content_digest
        and metadata.get("publication-sha256") == publication_digest
        and info.get("CacheControl") == cache_control
        and info.get("ContentType") == content_type
    )
 def publish_immutable_file(endpoint, bucket, key, source, cache_control, aws_env):
    """Publish a content-addressed key; identical retries converge."""
    content_type = mimetypes.guess_type(source.name)[0] or "application/octet-stream"
    content_digest, publication_digest = _immutable_digests(source, cache_control, content_type)
    address_digests = re.findall(r"(?<![0-9a-f])([0-9a-f]{64})(?![0-9a-f])", key.lower())
    if address_digests != [publication_digest]:
        raise RuntimeError(
            f"immutable key must contain its one publication SHA-256 {publication_digest}: "
            f"s3://{bucket}/{key}"
        )
    existing = _immutable_head(endpoint, bucket, key, aws_env)
    if existing is not None:
        if _same_immutable_object(
            existing, content_digest, publication_digest, cache_control, content_type,
        ):
            print(f"  Immutable object already matches: s3://{bucket}/{key}")
            return False
        raise RuntimeError(f"immutable object differs or lacks publisher digest: s3://{bucket}/{key}")
    args = ["aws", "--endpoint-url", endpoint, "s3api", "put-object",
            "--bucket", bucket, "--key", key, "--body", str(source),
            "--content-type", content_type, "--cache-control", cache_control,
            "--metadata", f"sha256={content_digest},publication-sha256={publication_digest}"]
    result = _aws_capture(args, aws_env)
    if result.returncode == 0:
        return True
    error = f"{result.stdout}\n{result.stderr}"
    raise RuntimeError(f"put-object failed for s3://{bucket}/{key}: {error.strip()}")
 def publish_immutable_rule(artifact, route, rule, html_dir, aws_env):
    """Publish one immutable cache partition without overwrite or deletion."""
    rule_root = html_dir / rule["path"]
    child_paths = [candidate["path"] for candidate in artifact["cache_rules"]
                   if candidate["path"].startswith(f"{rule['path'].rstrip('/')}/")]
    object_prefix = route["path"].strip("/")
    for source in sorted(path for path in rule_root.rglob("*") if path.is_file()):
        relative = source.relative_to(html_dir).as_posix()
        if any(relative == child or relative.startswith(f"{child}/") for child in child_paths):
            continue
        if any(fnmatch.fnmatch(relative, pattern) for pattern in artifact["excludes"]):
            continue
        key = "/".join(part for part in (object_prefix, relative) if part)
        publish_immutable_file(
            artifact["s3_endpoint"], artifact["bucket"], key, source,
            rule["cache_control"], aws_env,
        )
 def publication_aws_env(artifact, credential_env_names=None):
    """Build the route-scoped AWS environment without leaking other credentials."""
    access_key = env(artifact["credentials"]["access_key_env"])
    secret_key = env(artifact["credentials"]["secret_key_env"])
    aws_env = os.environ.copy()
@@ -167,160 +63,11 @@ def publication_aws_env(artifact, credential_env_names=None):
        "AWS_SECRET_ACCESS_KEY": secret_key,
        "AWS_DEFAULT_REGION": os.environ.get("AWS_DEFAULT_REGION", "sjc001"),
    })
    return aws_env
 def configure_cors(bucket, origins, endpoint, aws_env):
    """Reconcile read-only browser access without exposing publication credentials."""
    if origins is None:
        return
    config = None
    if origins:
        config = {
            "CORSRules": [{
                "AllowedOrigins": origins,
                "AllowedMethods": ["GET", "HEAD"],
                "AllowedHeaders": ["*"],
                "ExposeHeaders": ["ETag"],
                "MaxAgeSeconds": 3600,
            }],
        }
    set_cors_configuration(bucket, config, endpoint, aws_env)
 def set_cors_configuration(bucket, config, endpoint, aws_env):
    """Apply an exact bucket CORS configuration, or remove it when absent."""
    if config is None:
        run([
            "aws", "--endpoint-url", endpoint, "s3api", "delete-bucket-cors",
            "--bucket", bucket,
        ], env=aws_env)
        return
    with tempfile.NamedTemporaryFile("w", suffix=".json", encoding="utf-8") as handle:
        json.dump(config, handle)
        handle.flush()
        run([
            "aws", "--endpoint-url", endpoint, "s3api", "put-bucket-cors",
            "--bucket", bucket, "--cors-configuration", f"file://{handle.name}",
        ], env=aws_env)
 def get_cors_configuration(bucket, endpoint, aws_env):
    """Read the exact bucket CORS configuration for rollback."""
    result = _aws_capture([
        "aws", "--endpoint-url", endpoint, "s3api", "get-bucket-cors",
        "--bucket", bucket, "--output", "json",
    ], aws_env)
    if result.returncode == 0:
        try:
            config = json.loads(result.stdout)
        except json.JSONDecodeError as error:
            raise RuntimeError(f"get-bucket-cors returned invalid JSON for {bucket}") from error
        if not isinstance(config, dict) or not isinstance(config.get("CORSRules"), list):
            raise RuntimeError(f"get-bucket-cors returned an invalid policy for {bucket}")
        return config
    error = f"{result.stdout}\n{result.stderr}"
    if "NoSuchCORSConfiguration" in error:
        return None
    raise RuntimeError(f"get-bucket-cors failed for {bucket}: {error.strip()}")
 def reconcile_artifact_cors(artifacts, credential_env_names=None):
    """Reconcile all policies, restoring the prior set if any write fails."""
    snapshots = []
    for artifact in artifacts:
        if artifact["cors_origins"] is None:
            continue
        aws_env = publication_aws_env(artifact, credential_env_names)
        snapshots.append((
            artifact,
            aws_env,
            get_cors_configuration(artifact["bucket"], artifact["s3_endpoint"], aws_env),
        ))
    attempted = []
    try:
        for artifact, aws_env, previous in snapshots:
            attempted.append((artifact, aws_env, previous))
            configure_cors(
                artifact["bucket"], artifact["cors_origins"], artifact["s3_endpoint"], aws_env,
            )
    except Exception as error:
        rollback_errors = []
        for artifact, aws_env, previous in reversed(attempted):
            try:
                set_cors_configuration(
                    artifact["bucket"], previous, artifact["s3_endpoint"], aws_env,
                )
            except Exception as rollback_error:
                rollback_errors.append(f"{artifact['bucket']}: {rollback_error}")
        if rollback_errors:
            raise RuntimeError(
                f"CORS reconciliation failed ({error}); rollback also failed for "
                f"{'; '.join(rollback_errors)}"
            ) from error
        raise
 def publish_route_immutables(artifact, route, site_dir, credential_env_names=None):
    """Publish one route's immutable partitions during the global preflight."""
    html_dir = site_dir / artifact["build_dir"]
    aws_env = publication_aws_env(artifact, credential_env_names)
    for rule in artifact["cache_rules"]:
        if _is_immutable(rule):
            publish_immutable_rule(artifact, route, rule, html_dir, aws_env)
 def immutable_key_prefixes(artifact, route):
    """Return immutable partitions as bucket-relative key prefixes."""
    route_prefix = route["path"].strip("/")
    return [
        "/".join(part for part in (route_prefix, rule["path"]) if part)
        for rule in artifact["cache_rules"] if _is_immutable(rule)
    ]
 def retained_immutable_paths(artifact, route, previous_contract):
    """Carry all bucket history forward so later route moves cannot delete it."""
    previous_paths = previous_contract["immutable_paths"] if previous_contract else []
    return sorted(set(previous_paths) | set(immutable_key_prefixes(artifact, route)))
 def retired_immutable_filters(artifact, route, html_dir, previous_contract):
    """Protect historical immutable keys that fall inside the current sync scope."""
    if not previous_contract:
        return []
    current_prefix = route["path"].strip("/")
    filters = []
    current_immutable = set(immutable_key_prefixes(artifact, route))
    for immutable_path in previous_contract["immutable_paths"]:
        if immutable_path in current_immutable:
            continue
        if current_prefix:
            marker = f"{current_prefix}/"
            if not immutable_path.startswith(marker):
                continue
            retired_path = immutable_path[len(marker):]
        else:
            retired_path = immutable_path
        collision = html_dir / retired_path
        if collision.exists() and any(path.is_file() for path in collision.rglob("*")):
            raise RuntimeError(
                f"current artifact collides with retired immutable partition: {retired_path}"
            )
        filters.extend(("--exclude", f"{retired_path}/*"))
    return filters
 def s3_sync(artifact, route, site_dir, credential_env_names=None, previous_contract=None):
    endpoint = artifact["s3_endpoint"]
    html_dir = site_dir / artifact["build_dir"]
    aws_env = publication_aws_env(artifact, credential_env_names)
    bucket = artifact["bucket"]
    object_prefix = route["path"].strip("/")
-    destination = f"s3://{bucket}/{object_prefix + '/' if object_prefix else ''}"
+    bucket_destination = f"s3://{bucket}/"
    destination = f"{bucket_destination}{object_prefix + '/' if object_prefix else ''}"
    default_cache = next(rule["cache_control"] for rule in artifact["cache_rules"] if not rule["path"])
    immutable_paths = [rule["path"] for rule in artifact["cache_rules"] if _is_immutable(rule)]
    # `excludes` are patterns (site.yaml `excludes:` list) that should never
    # be uploaded *and* should never be deleted from the bucket — escape hatch
    # for assets managed out-of-band (e.g. large PDFs uploaded via aws-cli).
@@ -328,30 +75,41 @@ def s3_sync(artifact, route, site_dir, credential_env_names=None, previous_contr
    if artifact["excludes"]:
        print(f"Excluding patterns: {artifact['excludes']}")
    print(f"Syncing artifact {artifact['name']} → {destination} via {endpoint}")
-    # Upload with the final cache policy before cleanup. Sync and deletion are
+    # `sync --delete` handles new/changed/orphaned files. Partitioned
-    # scoped to the same current route prefix and cache partition. A route move
+    # `cp --recursive` calls then re-upload each file once to refresh metadata
-    # leaves its old bucket partition intact but unreachable after the old
+    # (cache-control, content-type) on objects sync skipped as unchanged.
    # Ingress disappears, while stale mutable keys on the serving prefix are
    # deleted. Immutable subtrees are structurally excluded. `cp --recursive`
    # refreshes metadata atomically per object before `sync --delete` removes
    # stale keys without ever exposing new bytes under a provisional policy.
    # A no-op deploy therefore transfers the artifact bytes once.
    # AWS CLI guesses Content-Type from file extension on local→S3 uploads,
    # so a fresh upload always carries the right MIME type.
    stage = None
    sync_source = html_dir
    sync_excludes = exclude_args
    if object_prefix:
        stage = tempfile.TemporaryDirectory()
        sync_source = Path(stage.name)
        staged_artifact = sync_source / object_prefix
        staged_artifact.parent.mkdir(parents=True, exist_ok=True)
        staged_artifact.symlink_to(html_dir.resolve(), target_is_directory=True)
        sync_excludes = [arg for pattern in artifact["excludes"]
                         for arg in ("--exclude", f"{object_prefix}/{pattern}")]
    try:
        # Sync the complete bucket authority so moving a route prefix also
        # deletes objects under its old prefix instead of leaving them public.
        run(["aws", "--endpoint-url", endpoint, "s3", "sync", f"{sync_source}/", bucket_destination,
             "--delete", "--only-show-errors", "--cache-control", default_cache,
             *sync_excludes], env=aws_env)
    finally:
        if stage:
            stage.cleanup()
    print("Re-stamping metadata on all objects...")
    specific_paths = [rule["path"] for rule in artifact["cache_rules"] if rule["path"]]
    default_filters = [arg for path in specific_paths for arg in ("--exclude", f"{path}/*")]
    retired_filters = retired_immutable_filters(artifact, route, html_dir, previous_contract)
    run(["aws", "--endpoint-url", endpoint, "s3", "cp", f"{html_dir}/", destination,
         "--recursive", "--only-show-errors", "--cache-control", default_cache,
-         *default_filters, *retired_filters, *exclude_args], env=aws_env)
+         *default_filters, *exclude_args], env=aws_env)
    run(["aws", "--endpoint-url", endpoint, "s3", "sync", f"{html_dir}/", destination,
         "--delete", "--only-show-errors", "--cache-control", default_cache,
         *default_filters, *retired_filters, *exclude_args], env=aws_env)
    for rule in artifact["cache_rules"]:
        if not rule["path"]:
            continue
        if _is_immutable(rule):
            continue
        include = f"{rule['path'].rstrip('/')}/*"
        child_filters = [arg for path in specific_paths
                         if path.startswith(f"{rule['path'].rstrip('/')}/")
@@ -361,9 +119,6 @@ def s3_sync(artifact, route, site_dir, credential_env_names=None, previous_contr
        run(["aws", "--endpoint-url", endpoint, "s3", "cp", f"{html_dir}/", destination,
             "--recursive", "--only-show-errors", "--cache-control", rule["cache_control"],
             "--exclude", "*", "--include", include, *child_filters, *exclude_args], env=aws_env)
        run(["aws", "--endpoint-url", endpoint, "s3", "sync", f"{html_dir}/", destination,
             "--delete", "--only-show-errors", "--cache-control", rule["cache_control"],
             "--exclude", "*", "--include", include, *child_filters, *exclude_args], env=aws_env)
 def garage_admin(method, path, token, body=None):
@@ -412,32 +167,16 @@ def ensure_bucket_aliases(site_name, aliases, admin_token):
            raise
-def render_site_manifests(
+def render_site_manifests(site_name, action_dir, app_dir, manifests_dir, cfg):
    site_name, action_dir, app_dir, manifests_dir, cfg, previous_contracts=None,
 ):
    """Always re-render manifests from current site.yaml. Templates own
    domain + aliases, so changes propagate without manual edits."""
    manifests_dir.mkdir(parents=True, exist_ok=True)
    artifact_by_name = {artifact["name"]: artifact for artifact in cfg["artifacts"]}
    routes = []
    previous_contracts = previous_contracts or {}
    next_contracts = {bucket: dict(contract)
                      for bucket, contract in previous_contracts.items()}
    for route in cfg["routes"]:
        artifact = artifact_by_name[route["artifact"]]
        resource_name = k8s_name(site_name) if cfg["compatibility"] else f"{k8s_name(site_name)}-{route['name']}"
-        previous = previous_contracts.get(artifact["bucket"])
+        routes.append({**route, "resource_name": resource_name, "artifact_config": artifact})
        immutable_paths = retained_immutable_paths(artifact, route, previous)
        next_contracts[artifact["bucket"]] = {
            "path": route["path"],
            "access": "public" if route["access"] == "legacy" else route["access"],
            "artifact": route["artifact"],
            "immutable_paths": immutable_paths,
        }
        routes.append({
            **route, "resource_name": resource_name, "artifact_config": artifact,
            "immutable_paths_json": json.dumps(immutable_paths, separators=(",", ":")),
        })
    template_vars = {
        "site": site_name,
        "site_k8s": k8s_name(site_name),
@@ -448,89 +187,6 @@ def render_site_manifests(
        "routes": routes,
    }
    render_templates(action_dir, template_vars, app_dir, manifests_dir)
    if not cfg["compatibility"] or previous_contracts:
        (app_dir / "site-publish-history.yaml").write_text(yaml.safe_dump(
            {"version": 1, "buckets": next_contracts}, sort_keys=True,
        ))
 def _validate_route_contract(bucket, contract, path):
    expected = {"path", "access", "artifact", "immutable_paths"}
    if (not isinstance(bucket, str) or not bucket or not isinstance(contract, dict)
            or set(contract) != expected
            or contract.get("access") not in {"public", "protected"}
            or not isinstance(contract.get("path"), str)
            or not contract["path"].startswith("/")
            or not isinstance(contract.get("artifact"), str) or not contract["artifact"]
            or not isinstance(contract.get("immutable_paths"), list)
            or any(not isinstance(item, str) or not item or item.startswith("/")
                   or any(part in {"", ".", ".."} for part in item.split("/"))
                   for item in contract["immutable_paths"])):
        raise RuntimeError(f"invalid site-publish route history in {path}")
    return {
        "path": contract["path"], "access": contract["access"],
        "artifact": contract["artifact"],
        "immutable_paths": sorted(set(contract["immutable_paths"])),
    }
 def previous_route_contracts(app_dir):
    """Read bucket-keyed route history from generated Ingresses."""
    history_path = app_dir / "site-publish-history.yaml"
    if history_path.exists():
        document = yaml.safe_load(history_path.read_text())
        if (not isinstance(document, dict) or set(document) != {"version", "buckets"}
                or document["version"] != 1 or not isinstance(document["buckets"], dict)):
            raise RuntimeError(f"invalid site-publish route history in {history_path}")
        return {
            bucket: _validate_route_contract(bucket, contract, history_path)
            for bucket, contract in document["buckets"].items()
        }
    contracts = {}
    manifests = app_dir / "manifests"
    if not manifests.exists():
        return contracts
    for path in sorted(manifests.glob("ingress*.yaml")):
        document = yaml.safe_load(path.read_text()) or {}
        annotations = document.get("metadata", {}).get("annotations", {})
        artifact = annotations.get("site-publish.fritzlab.net/artifact")
        access = annotations.get("site-publish.fritzlab.net/access")
        bucket = annotations.get("site-publish.fritzlab.net/bucket")
        immutable_paths_json = annotations.get("site-publish.fritzlab.net/immutable-paths")
        route_path = annotations.get("site-publish.fritzlab.net/route-path")
        values = (artifact, access, bucket, immutable_paths_json, route_path)
        if all(value is None for value in values):
            continue
        if (not all(isinstance(value, str) for value in values)
                or access not in {"public", "protected"} or not route_path.startswith("/")):
            raise RuntimeError(f"invalid site-publish route history in {path}")
        try:
            immutable_paths = json.loads(immutable_paths_json)
        except json.JSONDecodeError as exc:
            raise RuntimeError(f"invalid site-publish route history in {path}") from exc
        if not isinstance(immutable_paths, list) or any(not isinstance(item, str) for item in immutable_paths):
            raise RuntimeError(f"invalid site-publish route history in {path}")
        if bucket in contracts:
            raise RuntimeError(f"duplicate site-publish route history for bucket {bucket}")
        contracts[bucket] = _validate_route_contract(bucket, {
            "path": route_path, "access": access, "artifact": artifact,
            "immutable_paths": immutable_paths,
        }, path)
    return contracts
 def validate_route_migrations(cfg, previous_contracts):
    artifacts = {artifact["name"]: artifact for artifact in cfg["artifacts"]}
    for route in cfg["routes"]:
        artifact = artifacts[route["artifact"]]
        previous = previous_contracts.get(artifact["bucket"])
        if (previous and previous["access"] == "protected"
                and route["access"] in {"public", "legacy"}
        ):
            raise RuntimeError(
                f"artifact {route['artifact']} cannot become public while reusing protected "
                f"bucket {artifact['bucket']}"
            )
 def deploy_static(site_name, site_dir, action_dir, token, cfg):
@@ -541,35 +197,18 @@ def deploy_static(site_name, site_dir, action_dir, token, cfg):
    validate_publication_environment(cfg)
    for artifact in cfg["artifacts"]:
        validate_artifact_output(site_dir, artifact)
    apps_dir = clone_apps(token)
    app_dir = apps_dir / "sjc001" / "websites" / site_name
    manifests_dir = app_dir / "manifests"
    previous_contracts = previous_route_contracts(app_dir)
    validate_route_migrations(cfg, previous_contracts)
    # Complete immutable work across the whole publication before any route's
    # mutable pointers can change. Partial immutable success is safe; mixing a
    # new route with an old route after a later immutable failure is not.
    for route in cfg["routes"]:
-        publish_route_immutables(
+        s3_sync(artifact_by_name[route["artifact"]], route, site_dir, credential_env_names)
            artifact_by_name[route["artifact"]], route, site_dir, credential_env_names,
        )
    # Reconcile every browser-read policy before publishing mutable content.
    # A CORS failure therefore cannot leave a new channel pointing at a release
    # whose cross-origin assets browsers cannot consume.
    reconcile_artifact_cors(cfg["artifacts"], credential_env_names)
    for route in cfg["routes"]:
        s3_sync(
            artifact_by_name[route["artifact"]], route, site_dir, credential_env_names,
            previous_contracts.get(artifact_by_name[route["artifact"]]["bucket"]),
        )
    if cfg["compatibility"]:
        ensure_bucket_aliases(site_name, cfg["aliases"], os.environ.get("GARAGE_ADMIN_TOKEN"))
-    render_site_manifests(
+    apps_dir = clone_apps(token)
-        site_name, action_dir, app_dir, manifests_dir, cfg, previous_contracts,
+    app_dir = apps_dir / "sjc001" / "websites" / site_name
-    )
+    manifests_dir = app_dir / "manifests"
-    commit_and_push(apps_dir, f"Deploy {site_name}", token)
+    render_site_manifests(site_name, action_dir, app_dir, manifests_dir, cfg)
    commit_and_push(apps_dir, f"Deploy {site_name}")
 def decommission(site_name, token, buckets=None):
@@ -580,7 +219,7 @@ def decommission(site_name, token, buckets=None):
        print(f"No manifests for {site_name} — nothing to remove")
        return
    shutil.rmtree(site_path)
-    commit_and_push(apps_dir, f"Decommission {site_name}", token)
+    commit_and_push(apps_dir, f"Decommission {site_name}")
    for bucket in buckets or [site_name]:
        print(f"Bucket {bucket} and its objects are NOT purged automatically.")
        print(f"  garage bucket delete {bucket} --yes")
@@ -600,5 +239,4 @@ def cmd_deploy():
        decommission(site_name, token, [artifact["bucket"] for artifact in cfg["artifacts"]])
        return
    validate_artifact_inputs(site_dir, cfg)
    deploy_static(site_name, site_dir, action_dir, token, cfg)
@@ -1,6 +1,5 @@
 """Shared utilities for the site-publish action."""
 import ipaddress
 import os
 import re
 import shutil
@@ -27,7 +26,6 @@ EXCLUDE_FILES = {
 VALID_TYPES = {"static", "hugo", "mkdocs"}
 NAME_RE = re.compile(r"^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$")
 ENV_RE = re.compile(r"^[A-Z_][A-Z0-9_]*$")
 ACCESS_KEY_ENV_RE = re.compile(r"^([A-Z][A-Z0-9_]*)_S3_ACCESS_KEY$")
 BUCKET_RE = re.compile(r"^[a-z0-9](?:[a-z0-9.-]{1,61}[a-z0-9])?$")
 MIDDLEWARE_RE = re.compile(r"^[A-Za-z0-9](?:[A-Za-z0-9_-]{0,61}[A-Za-z0-9])?$")
@@ -105,55 +103,6 @@ def _strings(value, label):
    return values
 def _cors_origins(value, label):
    origins = _list(value, label)
    if any(not isinstance(item, str) or not item for item in origins):
        raise ConfigError(f"{label} must be a list of non-empty strings")
    canonical = []
    for origin in origins:
        if origin == "*":
            canonical.append(origin)
            continue
        parsed = urlparse(origin)
        try:
            port = parsed.port
        except ValueError:
            raise ConfigError(f"{label} must contain '*' or canonical HTTPS origins") from None
        if parsed.scheme != "https" or not parsed.hostname or parsed.path or parsed.params or (
            parsed.query or parsed.fragment or parsed.username or parsed.password
        ):
            raise ConfigError(f"{label} must contain '*' or canonical HTTPS origins")
        try:
            address = ipaddress.ip_address(parsed.hostname)
        except ValueError:
            try:
                hostname = parsed.hostname.encode("idna").decode("ascii")
            except UnicodeError:
                raise ConfigError(
                    f"{label} must contain '*' or canonical HTTPS origins"
                ) from None
            try:
                _hostname(hostname, f"{label} hostname")
            except ConfigError:
                raise ConfigError(
                    f"{label} must contain '*' or canonical HTTPS origins"
                ) from None
        else:
            if getattr(address, "scope_id", None) is not None:
                raise ConfigError(
                    f"{label} must contain '*' or canonical HTTPS origins"
                )
            hostname = f"[{address.compressed}]" if address.version == 6 else address.compressed
        canonical.append(f"https://{hostname}{f':{port}' if port not in (None, 443) else ''}")
    if len(canonical) != len(set(canonical)):
        raise ConfigError(f"{label} must not contain duplicate canonical origins")
    if "*" in canonical and len(canonical) != 1:
        raise ConfigError(f"{label} wildcard must be the only origin")
    if origins != canonical:
        raise ConfigError(f"{label} must contain '*' or canonical HTTPS origins")
    return origins
 def _hostname(value, label):
    if not isinstance(value, str) or len(value) > 253 or value.endswith("."):
        raise ConfigError(f"{label} must be a lowercase DNS hostname without a trailing dot")
@@ -265,7 +214,6 @@ def _legacy_config(raw, site_name):
        "website_authority": "garage-s3.storage.svc.k8s.sjc001.fritzlab.net",
        "credentials": {"access_key_env": "AWS_ACCESS_KEY_ID", "secret_key_env": "AWS_SECRET_ACCESS_KEY"},
        "cache_rules": [{"path": "", "cache_control": DEFAULT_CACHE_CONTROL}],
        "cors_origins": None,
    }
    return {
        "version": 1,
@@ -285,16 +233,12 @@ def _legacy_config(raw, site_name):
 def _artifact(item, index):
    label = f"artifacts[{index}]"
    item = _mapping(item, label)
-    _known_keys(
+    _known_keys(item, {"name", "type", "content_dir", "tidy", "excludes", "publish", "cache"}, label)
        item,
        {"name", "type", "content_dir", "tidy", "excludes", "publish", "cache", "cors_origins"},
        label,
    )
    name = item.get("name")
    if not isinstance(name, str) or not NAME_RE.fullmatch(name):
        raise ConfigError(f"{label}.name must be a DNS label")
    publish = _mapping(item.get("publish"), f"{label}.publish")
-    _known_keys(publish, {"bucket", "credentials"}, f"{label}.publish")
+    _known_keys(publish, {"bucket", "endpoint", "website_authority", "credentials"}, f"{label}.publish")
    bucket = publish.get("bucket")
    if not isinstance(bucket, str) or not BUCKET_RE.fullmatch(bucket):
        raise ConfigError(f"{label}.publish.bucket is not a valid bucket name")
@@ -306,15 +250,6 @@ def _artifact(item, index):
        if not isinstance(value, str) or not ENV_RE.fullmatch(value):
            raise ConfigError(f"{label}.publish.credentials.{key} must name an environment variable")
        normalized_credentials[key] = value
    access_match = ACCESS_KEY_ENV_RE.fullmatch(normalized_credentials["access_key_env"])
    expected_secret = (
        f"{access_match.group(1)}_S3_SECRET_KEY" if access_match else None
    )
    if normalized_credentials["secret_key_env"] != expected_secret:
        raise ConfigError(
            f"{label}.publish.credentials must be a matched "
            "<NAME>_S3_ACCESS_KEY and <NAME>_S3_SECRET_KEY pair"
        )
    cache = _mapping(item.get("cache"), f"{label}.cache")
    _known_keys(cache, {"rules"}, f"{label}.cache")
    rules = _list(cache.get("rules"), f"{label}.cache.rules")
@@ -335,15 +270,10 @@ def _artifact(item, index):
    if "" not in paths:
        raise ConfigError(f"{label}.cache.rules must declare a '/' default")
    cache_rules.sort(key=lambda rule: (len(PurePosixPath(rule["path"]).parts), rule["path"]))
-    immutable_paths = [rule["path"] for rule in cache_rules
+    authority = _hostname(
-                       if "immutable" in {part.strip().lower().split("=", 1)[0]
+        publish.get("website_authority", f"{bucket}.{DEFAULT_WEBSITE_SUFFIX}"),
-                                          for part in rule["cache_control"].split(",")}]
+        f"{label}.publish.website_authority",
-    if "" in immutable_paths:
+    )
        raise ConfigError(f"{label}.cache.rules immutable paths must be narrower than '/'")
    for path in immutable_paths:
        if any(other.startswith(f"{path}/") for other in paths):
            raise ConfigError(f"{label}.cache.rules cannot nest another policy under immutable /{path}")
    authority = f"{bucket}.{DEFAULT_WEBSITE_SUFFIX}"
    if not isinstance(item.get("tidy", True), bool):
        raise ConfigError(f"{label}.tidy must be a boolean")
    return {
@@ -354,14 +284,10 @@ def _artifact(item, index):
        "excludes": _strings(item.get("excludes") or [], f"{label}.excludes"),
        "build_dir": f".site-publish/{name}/html",
        "bucket": bucket,
-        "s3_endpoint": DEFAULT_S3_ENDPOINT,
+        "s3_endpoint": _endpoint(publish.get("endpoint", DEFAULT_S3_ENDPOINT), f"{label}.publish.endpoint"),
        "website_authority": authority,
        "credentials": normalized_credentials,
        "cache_rules": cache_rules,
        "cors_origins": (
            _cors_origins(item["cors_origins"], f"{label}.cors_origins")
            if "cors_origins" in item else []
        ),
    }
@@ -448,10 +374,6 @@ def _validate_multi(cfg):
                raise ConfigError(f"protected route {route['name']} cannot use shared-cache max-age")
            if route["access"] == "public" and "private" in directives:
                raise ConfigError(f"public route {route['name']} cannot use private cache policy")
        if route["access"] == "protected" and "*" in artifact["cors_origins"]:
            raise ConfigError(
                f"protected route {route['name']} cannot allow wildcard CORS"
            )
    root = next(route for route in routes if route["path"] == "/")
    if any(route["access"] == "protected" for route in routes) and root["access"] == "public":
        raise ConfigError("a public '/' catch-all would expose unmatched protected content")
@@ -488,42 +410,6 @@ def normalize_site_config(raw, site_name):
    return cfg
 def validate_artifact_inputs(site_dir, cfg):
    """Reject source containment before a public or protected build starts."""
    if cfg["compatibility"]:
        return
    root = Path(site_dir).resolve()
    sources = []
    for artifact in cfg["artifacts"]:
        declared = root
        for component in Path(artifact["content_dir"]).parts:
            declared /= component
            if declared.is_symlink():
                raise ConfigError(
                    f"artifact {artifact['name']} content_dir contains symlink component: "
                    f"{declared.relative_to(root)}"
                )
        source = declared.resolve()
        if source != root and root not in source.parents:
            raise ConfigError(
                f"artifact {artifact['name']} content_dir resolves outside the repository"
            )
        if source.exists():
            symlink = next((path for path in source.rglob("*") if path.is_symlink()), None)
            if symlink is not None:
                raise ConfigError(
                    f"artifact {artifact['name']} build input contains symlink: "
                    f"{symlink.relative_to(root)}"
                )
        sources.append((artifact["name"], source))
    for index, (name, source) in enumerate(sources):
        for other_name, other_source in sources[index + 1:]:
            if source == other_source or source in other_source.parents or other_source in source.parents:
                raise ConfigError(
                    f"artifact build inputs overlap after resolution: {name} and {other_name}"
                )
 def parse_site_yaml(site_dir, site_name=None):
    path = Path(site_dir) / "site.yaml"
    if not path.exists():
@@ -550,7 +436,10 @@ def clone_apps(token):
    apps_dir = Path("/tmp/apps-deploy")
    if apps_dir.exists():
        shutil.rmtree(apps_dir)
-    clone_env = git_auth_env(token)
+    clone_env = os.environ.copy()
    clone_env["CI_BOT_TOKEN"] = token
    clone_env["GIT_ASKPASS"] = str(Path(__file__).with_name("git-askpass.sh"))
    clone_env["GIT_TERMINAL_PROMPT"] = "0"
    url = f"https://{user}@{GITEA_HOST}/{APPS_REPO}.git"
    run(["git", "clone", "--depth", "1", url, str(apps_dir)],
        display=f"git clone --depth 1 https://{user}@{GITEA_HOST}/{APPS_REPO}.git {apps_dir}", env=clone_env)
@@ -588,23 +477,13 @@ def render_templates(action_dir, template_vars, app_dir, manifests_dir):
        print(f"  Rendered {out_name}.j2 -> {destination}")
-def git_auth_env(token):
+def commit_and_push(apps_dir, message):
    """Return credential-safe Git authentication shared by clone and push."""
    auth_env = os.environ.copy()
    auth_env["CI_BOT_TOKEN"] = token
    auth_env["GIT_ASKPASS"] = str(Path(__file__).with_name("git-askpass.sh"))
    auth_env["GIT_TERMINAL_PROMPT"] = "0"
    return auth_env
 def commit_and_push(apps_dir, message, token=None):
    run(["git", "-C", str(apps_dir), "add", "-A"])
    result = subprocess.run(["git", "-C", str(apps_dir), "diff", "--cached", "--quiet"], check=False)
    if result.returncode == 0:
        print("No manifest changes to commit")
        return False
    run(["git", "-C", str(apps_dir), "commit", "-m", message])
-    push_env = git_auth_env(token) if token else None
+    run(["git", "-C", str(apps_dir), "push"])
    run(["git", "-C", str(apps_dir), "push"], env=push_env)
    print("Manifests pushed — ArgoCD will sync")
    return True
@@ -4,13 +4,6 @@ metadata:
  name: {{ route.resource_name }}
  namespace: {{ namespace }}
  annotations:
 {%- if not compatibility %}
    site-publish.fritzlab.net/artifact: {{ route.artifact | tojson }}
    site-publish.fritzlab.net/access: {{ route.access | tojson }}
    site-publish.fritzlab.net/bucket: {{ route.artifact_config.bucket | tojson }}
    site-publish.fritzlab.net/immutable-paths: {{ route.immutable_paths_json | tojson }}
    site-publish.fritzlab.net/route-path: {{ route.path | tojson }}
 {%- endif %}
    traefik.ingress.kubernetes.io/router.middlewares: https-redirect@file,retry-upstream@file{% if route.access_middleware %},{{ route.access_middleware }}@file{% endif %}{% for m in route.middlewares %},{{ m }}@file{% endfor %}
 spec:
  ingressClassName: traefik
@@ -15,7 +15,6 @@ artifacts:
  - name: distributions
    type: static
    content_dir: dist
    cors_origins: ['*']
    publish:
      bucket: baseline-dist
      credentials:
@@ -1,19 +1,11 @@
 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
@@ -24,72 +16,13 @@ sys.path.insert(0, str(ROOT / "scripts"))
 import deploy
 import build
 import utils
-from utils import ConfigError, normalize_site_config, validate_artifact_inputs
+from utils import ConfigError, normalize_site_config
 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")
@@ -113,69 +46,6 @@ class ConfigContractTests(unittest.TestCase):
        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({
@@ -194,7 +64,6 @@ class ConfigContractTests(unittest.TestCase):
            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"
            )
@@ -215,11 +84,12 @@ class ConfigContractTests(unittest.TestCase):
        )
    def test_publication_credentials_cannot_be_reused(self):
-        def mutate(raw):
+        self.assert_invalid(
-            raw["artifacts"][0]["publish"]["credentials"] = copy.deepcopy(
+            lambda raw: raw["artifacts"][0]["publish"]["credentials"].__setitem__(
-                raw["artifacts"][1]["publish"]["credentials"]
+                "access_key_env", "DIST_S3_ACCESS_KEY"
            ),
            "publication credential DIST_S3_ACCESS_KEY is reused",
        )
        self.assert_invalid(mutate, "publication credential DIST_S3_ACCESS_KEY is reused")
    def test_cache_directive_contradiction_is_rejected(self):
        self.assert_invalid(
@@ -229,14 +99,6 @@ class ConfigContractTests(unittest.TestCase):
            "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__(
@@ -254,56 +116,6 @@ class ConfigContractTests(unittest.TestCase):
            "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 <NAME>_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):
@@ -327,7 +139,6 @@ class GenerationTests(unittest.TestCase):
            "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))
@@ -338,17 +149,6 @@ class GenerationTests(unittest.TestCase):
        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)
@@ -372,14 +172,12 @@ class GenerationTests(unittest.TestCase):
            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"))
+        tmp, _, 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):
@@ -405,75 +203,6 @@ class BuildTests(unittest.TestCase):
 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"
@@ -486,56 +215,6 @@ class PublishingTests(unittest.TestCase):
            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")
@@ -547,273 +226,31 @@ class PublishingTests(unittest.TestCase):
            (html / "channels").mkdir()
            (html / "releases" / "1.0.js").write_text("release")
            (html / "channels" / "stable.json").write_text("channel")
-            commands, events = [], []
+            commands = []
            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), \
+            }, clear=False), patch.object(deploy, "run", side_effect=capture), redirect_stdout(output):
                    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("s3://baseline-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("releases/" in command and
                                "public, max-age=31536000, immutable" in command
                                for command in rendered))
            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")
@@ -832,87 +269,5 @@ class PublishingTests(unittest.TestCase):
                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()