File size: 3,924 Bytes
bb02ad0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 | #!/usr/bin/env python3
"""Pull-based audit archiver: copy immutable round artifacts to the archive bucket.
Copies every ``manifests/round-*.json`` and ``receipts/**/round-*.json`` object
from the production bucket into an archive bucket, skipping keys that already
exist there (append-only, enforced client-side — Hippius has no WORM
primitive). Mutable pointers (``latest.json``) are deliberately NOT archived.
The archive bucket MUST live under a SEPARATE Hippius account: Hippius S3 keys
are account-scoped, so an archive under the production account would be
deletable by the same credentials it is meant to survive. Run this from a box
that holds ONLY the archive account's write keys and (optionally) production
read keys — never the reverse.
Environment:
SOURCE_S3_ENDPOINT default https://s3.hippius.com
SOURCE_S3_BUCKET e.g. cascade-testnet-manifests
SOURCE_S3_ACCESS_KEY / SOURCE_S3_SECRET_KEY (read)
ARCHIVE_S3_ENDPOINT default https://s3.hippius.com
ARCHIVE_S3_BUCKET e.g. cascade-mainnet-archive
ARCHIVE_S3_ACCESS_KEY / ARCHIVE_S3_SECRET_KEY (write — the ONLY place these live)
Usage:
archive_sync.py [--dry-run] [--prefixes manifests/ receipts/]
Cron (hourly is plenty — rounds are epoch-paced):
17 * * * * cd /opt/cascade-archiver && ./archive_sync.py >> archive.log 2>&1
"""
from __future__ import annotations
import argparse
import fnmatch
import os
import sys
import boto3
from botocore.config import Config
IMMUTABLE_PATTERNS = ("manifests/round-*.json", "receipts/*/round-*.json",
"receipts/round-*.json")
CFG = Config(connect_timeout=10, read_timeout=60,
retries={"max_attempts": 3, "mode": "standard"})
def _client(side: str):
endpoint = os.environ.get(f"{side}_S3_ENDPOINT", "https://s3.hippius.com")
return boto3.client(
"s3", endpoint_url=endpoint,
aws_access_key_id=os.environ[f"{side}_S3_ACCESS_KEY"],
aws_secret_access_key=os.environ[f"{side}_S3_SECRET_KEY"],
config=CFG,
)
def _list_keys(s3, bucket: str, prefix: str) -> set[str]:
keys: set[str] = set()
token: str | None = None
while True:
kw = {"Bucket": bucket, "Prefix": prefix}
if token:
kw["ContinuationToken"] = token
resp = s3.list_objects_v2(**kw)
keys.update(o["Key"] for o in resp.get("Contents", []))
if not resp.get("IsTruncated"):
return keys
token = resp.get("NextContinuationToken")
def main(argv: list[str] | None = None) -> int:
p = argparse.ArgumentParser(description=__doc__)
p.add_argument("--dry-run", action="store_true",
help="list what would be copied; write nothing")
p.add_argument("--prefixes", nargs="+", default=["manifests/", "receipts/"])
args = p.parse_args(argv)
src = _client("SOURCE")
src_bucket = os.environ["SOURCE_S3_BUCKET"]
dst = None
dst_bucket = os.environ.get("ARCHIVE_S3_BUCKET", "")
if not args.dry_run:
dst = _client("ARCHIVE")
copied = skipped = 0
for prefix in args.prefixes:
source_keys = {
k for k in _list_keys(src, src_bucket, prefix)
if any(fnmatch.fnmatch(k, pat) for pat in IMMUTABLE_PATTERNS)
}
have = _list_keys(dst, dst_bucket, prefix) if dst is not None else set()
for key in sorted(source_keys - have):
if args.dry_run:
print(f"WOULD COPY {key}")
copied += 1
continue
body = src.get_object(Bucket=src_bucket, Key=key)["Body"].read()
dst.put_object(Bucket=dst_bucket, Key=key, Body=body)
print(f"archived {key} ({len(body)} bytes)")
copied += 1
skipped += len(source_keys & have)
print(f"done: {copied} copied, {skipped} already archived")
return 0
if __name__ == "__main__":
sys.exit(main())
|