fix: make CORS reconciliation recoverable

Authored-By: @architect <architect@fritzlab.net>

README.md

@@ -52,6 +52,7 @@ artifacts:
   - name: distributions
     type: static
     content_dir: dist
+    cors_origins: ['*']
     publish:
       bucket: baseline-dist
       credentials:
@@ -92,8 +93,10 @@ routes:
 ```
 
 The caller supplies each declared credential name as an environment variable
-on the action step. Credential values are passed to `aws` only through its
-environment and never appear in a logged command or process argument.
+on the action step. Names must be matched `<NAME>_S3_ACCESS_KEY` and
+`<NAME>_S3_SECRET_KEY` pairs; arbitrary environment variables cannot become
+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
@@ -120,6 +123,40 @@ 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

scripts/build.py

@@ -5,7 +5,7 @@ import subprocess
 import tempfile
 from pathlib import Path
 
-from utils import EXCLUDE_FILES, env, parse_site_yaml, run
+from utils import EXCLUDE_FILES, env, parse_site_yaml, run, validate_artifact_inputs
 
 
 def build_artifact(site_dir, artifact):
@@ -64,5 +64,7 @@ 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)

scripts/deploy.py

@@ -1,13 +1,20 @@
 """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,
@@ -18,6 +25,7 @@ from utils import (
     parse_site_yaml,
     render_templates,
     run,
+    validate_artifact_inputs,
 )
 
 GARAGE_ADMIN_ENDPOINT = os.environ.get(
@@ -47,9 +55,105 @@ def validate_publication_environment(cfg):
         die("GARAGE_ADMIN_TOKEN is required when aliases are declared")
 
 
-def s3_sync(artifact, route, site_dir, credential_env_names=None):
-    endpoint = artifact["s3_endpoint"]
-    html_dir = site_dir / artifact["build_dir"]
+def _is_immutable(rule):
+    return "immutable" in {
+        part.strip().lower().split("=", 1)[0]
+        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()
@@ -63,11 +167,160 @@ def s3_sync(artifact, route, site_dir, 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("/")
-    bucket_destination = f"s3://{bucket}/"
-    destination = f"{bucket_destination}{object_prefix + '/' if object_prefix else ''}"
+    destination = f"s3://{bucket}/{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).
@@ -75,41 +328,30 @@ def s3_sync(artifact, route, site_dir, credential_env_names=None):
     if artifact["excludes"]:
         print(f"Excluding patterns: {artifact['excludes']}")
     print(f"Syncing artifact {artifact['name']} → {destination} via {endpoint}")
-    # `sync --delete` handles new/changed/orphaned files. Partitioned
-    # `cp --recursive` calls then re-upload each file once to refresh metadata
-    # (cache-control, content-type) on objects sync skipped as unchanged.
+    # Upload with the final cache policy before cleanup. Sync and deletion are
+    # scoped to the same current route prefix and cache partition. A route move
+    # leaves its old bucket partition intact but unreachable after the old
+    # 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, *exclude_args], env=aws_env)
+         *default_filters, *retired_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('/')}/")
@@ -119,6 +361,9 @@ def s3_sync(artifact, route, site_dir, credential_env_names=None):
         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):
@@ -167,16 +412,32 @@ def ensure_bucket_aliases(site_name, aliases, admin_token):
             raise
 
 
-def render_site_manifests(site_name, action_dir, app_dir, manifests_dir, cfg):
+def render_site_manifests(
+    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']}"
-        routes.append({**route, "resource_name": resource_name, "artifact_config": artifact})
+        previous = previous_contracts.get(artifact["bucket"])
+        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),
@@ -187,6 +448,89 @@ def render_site_manifests(site_name, action_dir, app_dir, manifests_dir, cfg):
         "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):
@@ -197,18 +541,35 @@ 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)
-    for route in cfg["routes"]:
-        s3_sync(artifact_by_name[route["artifact"]], route, site_dir, credential_env_names)
-    if cfg["compatibility"]:
-        ensure_bucket_aliases(site_name, cfg["aliases"], os.environ.get("GARAGE_ADMIN_TOKEN"))
-
     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(
+            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(site_name, action_dir, app_dir, manifests_dir, cfg)
+    render_site_manifests(
+        site_name, action_dir, app_dir, manifests_dir, cfg, previous_contracts,
+    )
 
-    commit_and_push(apps_dir, f"Deploy {site_name}")
+    commit_and_push(apps_dir, f"Deploy {site_name}", token)
 
 
 def decommission(site_name, token, buckets=None):
@@ -219,7 +580,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}")
+    commit_and_push(apps_dir, f"Decommission {site_name}", token)
     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")
@@ -239,4 +600,5 @@ 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)

scripts/utils.py

@@ -1,5 +1,6 @@
 """Shared utilities for the site-publish action."""
 
+import ipaddress
 import os
 import re
 import shutil
@@ -26,6 +27,7 @@ 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])?$")
 
@@ -103,6 +105,55 @@ 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")
@@ -214,6 +265,7 @@ 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,
@@ -233,12 +285,16 @@ def _legacy_config(raw, site_name):
 def _artifact(item, index):
     label = f"artifacts[{index}]"
     item = _mapping(item, label)
-    _known_keys(item, {"name", "type", "content_dir", "tidy", "excludes", "publish", "cache"}, label)
+    _known_keys(
+        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", "endpoint", "website_authority", "credentials"}, f"{label}.publish")
+    _known_keys(publish, {"bucket", "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")
@@ -250,6 +306,15 @@ 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")
@@ -270,10 +335,15 @@ 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"]))
-    authority = _hostname(
-        publish.get("website_authority", f"{bucket}.{DEFAULT_WEBSITE_SUFFIX}"),
-        f"{label}.publish.website_authority",
-    )
+    immutable_paths = [rule["path"] for rule in cache_rules
+                       if "immutable" in {part.strip().lower().split("=", 1)[0]
+                                          for part in rule["cache_control"].split(",")}]
+    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 {
@@ -284,10 +354,14 @@ def _artifact(item, index):
         "excludes": _strings(item.get("excludes") or [], f"{label}.excludes"),
         "build_dir": f".site-publish/{name}/html",
         "bucket": bucket,
-        "s3_endpoint": _endpoint(publish.get("endpoint", DEFAULT_S3_ENDPOINT), f"{label}.publish.endpoint"),
+        "s3_endpoint": DEFAULT_S3_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 []
+        ),
     }
 
 
@@ -374,6 +448,10 @@ 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")
@@ -410,6 +488,42 @@ 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():
@@ -436,10 +550,7 @@ def clone_apps(token):
     apps_dir = Path("/tmp/apps-deploy")
     if apps_dir.exists():
         shutil.rmtree(apps_dir)
-    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"
+    clone_env = git_auth_env(token)
     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)
@@ -477,13 +588,23 @@ def render_templates(action_dir, template_vars, app_dir, manifests_dir):
         print(f"  Rendered {out_name}.j2 -> {destination}")
 
 
-def commit_and_push(apps_dir, message):
+def git_auth_env(token):
+    """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])
-    run(["git", "-C", str(apps_dir), "push"])
+    push_env = git_auth_env(token) if token else None
+    run(["git", "-C", str(apps_dir), "push"], env=push_env)
     print("Manifests pushed — ArgoCD will sync")
     return True

templates/ingress.yaml.j2

@@ -4,6 +4,13 @@ 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

tests/fixtures/split-site.yaml

@@ -15,6 +15,7 @@ artifacts:
   - name: distributions
     type: static
     content_dir: dist
+    cors_origins: ['*']
     publish:
       bucket: baseline-dist
       credentials:

tests/test_contract.py

@@ -1,11 +1,19 @@
+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
@@ -16,13 +24,72 @@ sys.path.insert(0, str(ROOT / "scripts"))
 import deploy
 import build
 import utils
-from utils import ConfigError, normalize_site_config
+from utils import ConfigError, normalize_site_config, 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")
@@ -46,6 +113,69 @@ 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({
@@ -64,6 +194,7 @@ 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"
             )
@@ -84,12 +215,11 @@ class ConfigContractTests(unittest.TestCase):
         )
 
     def test_publication_credentials_cannot_be_reused(self):
-        self.assert_invalid(
-            lambda raw: raw["artifacts"][0]["publish"]["credentials"].__setitem__(
-                "access_key_env", "DIST_S3_ACCESS_KEY"
-            ),
-            "publication credential DIST_S3_ACCESS_KEY is reused",
+        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(
@@ -99,6 +229,14 @@ 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__(
@@ -116,6 +254,56 @@ 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):
@@ -139,6 +327,7 @@ 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))
@@ -149,6 +338,17 @@ 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)
@@ -172,12 +372,14 @@ class GenerationTests(unittest.TestCase):
             self.assertFalse(stale.exists())
 
     def test_legacy_names_and_garage_s3_target_are_preserved(self):
-        tmp, _, files = self.render(fixture("legacy-site.yaml"))
+        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):
@@ -203,6 +405,75 @@ 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"
@@ -215,6 +486,56 @@ 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")
@@ -226,31 +547,273 @@ class PublishingTests(unittest.TestCase):
             (html / "channels").mkdir()
             (html / "releases" / "1.0.js").write_text("release")
             (html / "channels" / "stable.json").write_text("channel")
-            commands = []
+            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), redirect_stdout(output):
+            }, 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/", rendered[0])
+            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("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")
@@ -269,5 +832,87 @@ 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()