Warden v3

.github/scripts/copy-private-artifacts.py

@@ -0,0 +1,136 @@
+#!/usr/bin/env python3
+"""Copy version artifacts out of a public R2 bucket, verifying both byte streams.
+
+This never deletes source objects or changes CDN configuration. AWS CLI credentials
+and endpoint come from the operator's environment; no credentials are logged.
+"""
+
+import argparse
+import hashlib
+import json
+import os
+import subprocess
+import tempfile
+from pathlib import Path
+
+
+PREFIXES = ("files/", "modpack-overrides/", "modpacks/")
+MAX_OBJECT_BYTES = 1024 * 1024 * 1024
+
+
+def aws(endpoint, *args):
+    command = ["aws", "--endpoint-url", endpoint, "s3api", *args, "--output", "json", "--no-cli-pager"]
+    result = subprocess.run(command, check=True, text=True, capture_output=True)
+    return json.loads(result.stdout) if result.stdout.strip() else {}
+
+
+def inventory(endpoint, bucket):
+    seen = set()
+    for prefix in PREFIXES:
+        continuation = None
+        while True:
+            args = ["list-objects-v2", "--bucket", bucket, "--prefix", prefix,
+                    "--max-keys", "1000", "--no-paginate"]
+            if continuation:
+                args.extend(("--continuation-token", continuation))
+            page = aws(endpoint, *args)
+            for item in page.get("Contents", []):
+                key = item["Key"]
+                if not key.startswith(prefix) or key in seen:
+                    raise ValueError("Unexpected or duplicate artifact key in source inventory")
+                seen.add(key)
+                size = item["Size"]
+                if not isinstance(size, int) or size < 0 or size > MAX_OBJECT_BYTES:
+                    raise ValueError("Artifact exceeds migration byte limit")
+                yield key, size
+            if not page.get("IsTruncated"):
+                break
+            next_token = page.get("NextContinuationToken")
+            if not next_token or next_token == continuation:
+                raise ValueError("Artifact inventory pagination did not advance")
+            continuation = next_token
+
+
+def download(endpoint, bucket, key, path, expected_size):
+    metadata = aws(endpoint, "get-object", "--bucket", bucket, "--key", key, str(path))
+    actual_size = path.stat().st_size
+    if actual_size != expected_size or metadata.get("ContentLength") != expected_size:
+        raise ValueError("Artifact size changed during migration")
+    with path.open("rb") as source:
+        digest = hashlib.file_digest(source, "sha256").hexdigest()
+    return metadata, digest
+
+
+def destination_exists(endpoint, bucket, key):
+    try:
+        aws(endpoint, "head-object", "--bucket", bucket, "--key", key)
+        return True
+    except subprocess.CalledProcessError as error:
+        # AWS CLI emits its diagnostic on stderr. Only an absent object may be copied.
+        if "404" in error.stderr or "Not Found" in error.stderr or "NoSuchKey" in error.stderr:
+            return False
+        raise
+
+
+def copy_and_verify(endpoint, source, destination, key, size, workdir):
+    source_file = workdir / "source"
+    destination_file = workdir / "destination"
+    source_metadata, source_digest = download(endpoint, source, key, source_file, size)
+    already_present = destination_exists(endpoint, destination, key)
+    if not already_present:
+        args = ["put-object", "--bucket", destination, "--key", key, "--body", str(source_file)]
+        args.extend(("--cache-control", "private, no-store"))
+        for field, flag in (("ContentType", "--content-type"), ("ContentDisposition", "--content-disposition")):
+            if source_metadata.get(field):
+                args.extend((flag, source_metadata[field]))
+        aws(endpoint, *args)
+    _, destination_digest = download(endpoint, destination, key, destination_file, size)
+    if destination_digest != source_digest:
+        raise ValueError("Private artifact bytes do not match source")
+    source_file.unlink()
+    destination_file.unlink()
+    return already_present, source_digest
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--endpoint", required=True)
+    parser.add_argument("--source-bucket", required=True)
+    parser.add_argument("--destination-bucket", required=True)
+    parser.add_argument("--copy", action="store_true", help="Copy and verify every object; default only inventories")
+    parser.add_argument("--report", type=Path, required=True, help="Write an atomic summary without artifact keys")
+    args = parser.parse_args()
+    if args.source_bucket == args.destination_bucket:
+        parser.error("Source and destination buckets must differ")
+    if not args.endpoint.startswith("https://"):
+        parser.error("An HTTPS R2 endpoint is required")
+    count = 0
+    total_bytes = 0
+    copied = 0
+    already_equal = 0
+    manifest_digest = hashlib.sha256()
+    with tempfile.TemporaryDirectory(prefix="modtale-artifact-copy-") as directory:
+        for key, size in inventory(args.endpoint, args.source_bucket):
+            count += 1
+            total_bytes += size
+            if args.copy:
+                existed, digest = copy_and_verify(args.endpoint, args.source_bucket,
+                                                  args.destination_bucket, key, size, Path(directory))
+                already_equal += int(existed)
+                copied += int(not existed)
+                manifest_digest.update(key.encode("utf-8") + b"\0" + digest.encode("ascii") + b"\n")
+    report = {"sourceBucket": args.source_bucket, "destinationBucket": args.destination_bucket,
+              "objectCount": count, "totalBytes": total_bytes, "copied": copied,
+              "alreadyEqual": already_equal, "copyVerified": args.copy,
+              "keyAndByteManifestSha256": manifest_digest.hexdigest() if args.copy else None}
+    args.report.parent.mkdir(parents=True, exist_ok=True)
+    with tempfile.NamedTemporaryFile(mode="w", dir=args.report.parent, delete=False) as file:
+        json.dump(report, file, sort_keys=True, indent=2)
+        file.write("\n")
+        temporary = file.name
+    os.replace(temporary, args.report)
+    print(json.dumps(report, sort_keys=True))
+
+
+if __name__ == "__main__":
+    main()

.github/scripts/verify-private-artifact-edge.py

@@ -0,0 +1,90 @@
+#!/usr/bin/env python3
+"""Read-only R2/CDN cutover check. Requires CLOUDFLARE_API_TOKEN and known sample keys."""
+
+import argparse
+import json
+import os
+import urllib.error
+import urllib.parse
+import urllib.request
+
+
+PREFIXES = ("files/", "modpack-overrides/", "modpacks/")
+
+
+def cloudflare(account, bucket, route, token):
+    path = "/client/v4/accounts/{}/r2/buckets/{}/domains/{}".format(
+        urllib.parse.quote(account, safe=""), urllib.parse.quote(bucket, safe=""), route)
+    request = urllib.request.Request("https://api.cloudflare.com" + path,
+                                     headers={"Authorization": "Bearer " + token})
+    with urllib.request.urlopen(request, timeout=15) as response:
+        result = json.load(response)
+    if result.get("success") is not True or not isinstance(result.get("result"), dict):
+        raise ValueError("Cloudflare did not return a valid domain configuration")
+    return result["result"]
+
+
+def domains_are_private(account, public_bucket, private_bucket, cdn_host, token):
+    for bucket in (public_bucket, private_bucket):
+        managed = cloudflare(account, bucket, "managed", token)
+        if managed.get("enabled") is not False:
+            raise ValueError("An R2-managed public endpoint is enabled or unverified")
+    public_domains = cloudflare(account, public_bucket, "custom", token).get("domains")
+    private_domains = cloudflare(account, private_bucket, "custom", token).get("domains")
+    if not isinstance(public_domains, list) or not isinstance(private_domains, list):
+        raise ValueError("Cloudflare custom-domain inventory is incomplete")
+    if any(domain.get("enabled") is not False for domain in private_domains):
+        raise ValueError("Private artifact bucket has a public custom domain")
+    active = [domain.get("domain") for domain in public_domains if domain.get("enabled") is True]
+    if active != [cdn_host]:
+        raise ValueError("Public bucket has unexpected enabled custom domains")
+
+
+def probe(cdn_host, key, method, headers=None):
+    url = "https://" + cdn_host + "/" + urllib.parse.quote(key, safe="/-_.")
+    request = urllib.request.Request(url, method=method, headers=headers or {})
+    try:
+        with urllib.request.urlopen(request, timeout=15) as response:
+            return response.status
+    except urllib.error.HTTPError as error:
+        return error.code
+
+
+def verify_samples(cdn_host, artifact_keys, media_key):
+    if {next((prefix for prefix in PREFIXES if key.startswith(prefix)), None)
+            for key in artifact_keys} != set(PREFIXES):
+        raise ValueError("Provide one sample artifact key for each protected prefix")
+    if any(media_key.startswith(prefix) for prefix in PREFIXES):
+        raise ValueError("Media sample must not use an artifact prefix")
+    for key in artifact_keys:
+        if probe(cdn_host, key, "HEAD") != 403:
+            raise ValueError("CDN did not deny an artifact HEAD request")
+        if probe(cdn_host, key, "GET", {"Range": "bytes=0-0"}) != 403:
+            raise ValueError("CDN did not deny an artifact byte-range request")
+    if probe(cdn_host, media_key, "HEAD") != 200:
+        raise ValueError("Public media did not remain available")
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--account-id", required=True)
+    parser.add_argument("--public-bucket", required=True)
+    parser.add_argument("--private-bucket", required=True)
+    parser.add_argument("--cdn-host", required=True)
+    parser.add_argument("--artifact-key", action="append", required=True)
+    parser.add_argument("--media-key", required=True)
+    args = parser.parse_args()
+    token = os.environ.get("CLOUDFLARE_API_TOKEN")
+    if not token:
+        parser.error("CLOUDFLARE_API_TOKEN is required")
+    if args.public_bucket == args.private_bucket:
+        parser.error("Buckets must differ")
+    if args.cdn_host != "cdn.modtale.net":
+        parser.error("The production CDN host must be checked explicitly")
+    domains_are_private(args.account_id, args.public_bucket, args.private_bucket, args.cdn_host, token)
+    verify_samples(args.cdn_host, args.artifact_key, args.media_key)
+    print("Private artifact domain settings and CDN sample denials verified")
+
+
+if __name__ == "__main__":
+    main()

.github/tests/copy-private-artifacts.test.py

@@ -0,0 +1,80 @@
+import importlib.util
+import hashlib
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+
+SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "copy-private-artifacts.py"
+spec = importlib.util.spec_from_file_location("private_artifact_copy", SCRIPT)
+module = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(module)
+
+
+class PrivateArtifactCopyTest(unittest.TestCase):
+    def test_inventory_paginates_all_artifact_prefixes_and_rejects_stalls(self):
+        def aws(_endpoint, *args):
+            prefix = args[args.index("--prefix") + 1]
+            if prefix == "files/" and "--continuation-token" not in args:
+                return {"Contents": [{"Key": "files/mod/a.jar", "Size": 3}],
+                        "IsTruncated": True, "NextContinuationToken": "next"}
+            if prefix == "files/":
+                return {"Contents": [{"Key": "files/mod/b.jar", "Size": 4}], "IsTruncated": False}
+            return {"Contents": [], "IsTruncated": False}
+
+        with patch.object(module, "aws", side_effect=aws):
+            self.assertEqual([("files/mod/a.jar", 3), ("files/mod/b.jar", 4)],
+                             list(module.inventory("https://r2.test", "public")))
+        with patch.object(module, "aws", return_value={"Contents": [], "IsTruncated": True}):
+            with self.assertRaisesRegex(ValueError, "did not advance"):
+                list(module.inventory("https://r2.test", "public"))
+
+    def test_copy_verifies_destination_bytes_and_preserves_source(self):
+        calls = []
+
+        def aws(_endpoint, *args):
+            calls.append(args)
+            if args[0] == "head-object":
+                return {}
+            if args[0] == "get-object":
+                bucket = args[args.index("--bucket") + 1]
+                Path(args[-1]).write_bytes(b"abc" if bucket == "public" else b"bad")
+                return {"ContentLength": 3}
+            raise AssertionError(args)
+
+        with tempfile.TemporaryDirectory() as directory, patch.object(module, "aws", side_effect=aws):
+            with self.assertRaisesRegex(ValueError, "do not match"):
+                module.copy_and_verify("https://r2.test", "public", "private", "files/mod/a.jar", 3,
+                                       Path(directory))
+        self.assertFalse(any(args[0] == "delete-object" for args in calls))
+
+    def test_new_copy_preserves_bytes_and_sets_private_cache_policy(self):
+        stored = {}
+        calls = []
+
+        def aws(_endpoint, *args):
+            calls.append(args)
+            if args[0] == "get-object":
+                bucket = args[args.index("--bucket") + 1]
+                content = b"good" if bucket == "public" else stored["bytes"]
+                Path(args[-1]).write_bytes(content)
+                return {"ContentLength": len(content), "ContentType": "application/java-archive"}
+            if args[0] == "put-object":
+                stored["bytes"] = Path(args[args.index("--body") + 1]).read_bytes()
+                return {}
+            raise AssertionError(args)
+
+        with tempfile.TemporaryDirectory() as directory, patch.object(module, "aws", side_effect=aws), \
+                patch.object(module, "destination_exists", return_value=False):
+            existed, digest = module.copy_and_verify("https://r2.test", "public", "private",
+                                                     "files/mod/a.jar", 4, Path(directory))
+        self.assertFalse(existed)
+        self.assertEqual(b"good", stored["bytes"])
+        self.assertEqual(hashlib.sha256(b"good").hexdigest(), digest)
+        self.assertTrue(any(args[0] == "put-object" and
+                            args[args.index("--cache-control") + 1] == "private, no-store" for args in calls))
+
+
+if __name__ == "__main__":
+    unittest.main()

.github/tests/verify-private-artifact-edge.test.py

@@ -0,0 +1,50 @@
+import importlib.util
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+
+SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "verify-private-artifact-edge.py"
+spec = importlib.util.spec_from_file_location("artifact_edge_verify", SCRIPT)
+module = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(module)
+
+
+class ArtifactEdgeVerifyTest(unittest.TestCase):
+    def test_rejects_any_public_route_to_private_bucket_or_managed_public_endpoint(self):
+        def cloudflare(_account, bucket, route, _token):
+            if route == "managed":
+                return {"enabled": False}
+            return {"domains": [{"domain": "cdn.modtale.net", "enabled": True}]}
+
+        with patch.object(module, "cloudflare", side_effect=cloudflare):
+            with self.assertRaisesRegex(ValueError, "Private artifact bucket"):
+                module.domains_are_private("account", "public", "private", "cdn.modtale.net", "token")
+
+        def managed_bypass(_account, bucket, route, _token):
+            if route == "managed":
+                return {"enabled": bucket == "public"}
+            return {"domains": []}
+
+        with patch.object(module, "cloudflare", side_effect=managed_bypass):
+            with self.assertRaisesRegex(ValueError, "R2-managed"):
+                module.domains_are_private("account", "public", "private", "cdn.modtale.net", "token")
+
+    def test_requires_all_prefixes_denied_for_head_and_range_while_media_works(self):
+        keys = ["files/mod/a.jar", "modpack-overrides/b.zip", "modpacks/c.zip"]
+
+        def probe(_cdn, key, method, _headers=None):
+            return 200 if key == "images/media.png" else 403
+
+        with patch.object(module, "probe", side_effect=probe) as observed:
+            module.verify_samples("cdn.modtale.net", keys, "images/media.png")
+            self.assertEqual(7, observed.call_count)
+        with patch.object(module, "probe", return_value=200):
+            with self.assertRaisesRegex(ValueError, "HEAD"):
+                module.verify_samples("cdn.modtale.net", keys, "images/media.png")
+        with self.assertRaisesRegex(ValueError, "each protected prefix"):
+            module.verify_samples("cdn.modtale.net", keys[:2], "images/media.png")
+
+
+if __name__ == "__main__":
+    unittest.main()