Source identity
| Field | Recorded value |
|---|---|
| Source type | archive member |
| Original record | 03_recover_deploys.py |
| Date | Not specified in source |
| Path or archive member | archives/libsrc-1c3f88063121/NETLIFY_AI_RECOVERY_2026-10-14_21-10-49/scripts/03_recover_deploys.py |
| Original SHA-256 | 44d74f73d720538c8a40345e48014258ce7e54082dcc18df50de0e534bc29c78 |
| Reading-copy SHA-256 | 44d74f73d720538c8a40345e48014258ce7e54082dcc18df50de0e534bc29c78 |
| Archive source | {"sourceId": "libsrc-1c3f88063121", "member": "NETLIFY_AI_RECOVERY_2026-10-14_21-10-49/scripts/03_recover_deploys.py"} |
Recovered source transcript
Download complete source text ↓278 lines · 14.7 KB
"""Phase 3: inventory deploy history for included sites and recover deployed files, one deploy at a time.
Retrieval order per file (every byte stream is accepted only if its SHA1 equals Netlify's manifest sha):
1. Content already recovered in this site's content store (same sha1 means identical bytes).
2. Documented raw file endpoint GET /sites/{id}/files/{path} (current published deploy only), used when the
deploy is the published one or when the current deploy holds the same sha at some path.
3. The deploy's own permalink https://{deploy_id}--{site}.netlify.app/{path} (public, unauthenticated).
Files that fail all three are recorded as missing with the reason. Nothing is rewritten or regenerated.
Resumable: deploys marked verified in the checkpoint are skipped."""
import concurrent.futures as cf
import datetime
import hashlib
import json
import os
import sys
import unicodedata
import urllib.parse
import zipfile
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
from nlr_common import (API, ROOT, api_json, checkpoint_read, checkpoint_update, free_gb, log, now, paginate,
request, safe_name, sha1, sha256_file, write_bytes_atomic, write_json)
MIN_FREE_GB = 25
DEPLOY_FIELDS = ["id", "site_id", "state", "context", "branch", "commit_ref", "commit_url", "title", "created_at",
"updated_at", "published_at", "deploy_time", "deploy_ssl_url", "framework", "manual_deploy",
"deploy_source", "skipped", "locked", "expires_at", "has_source_zip", "edge_functions_present",
"function_schedules", "blobs_region", "review_id", "entry_path", "public_repo"]
STORED_EXT = (".png", ".jpg", ".jpeg", ".gif", ".webp", ".avif", ".woff", ".woff2", ".zip", ".gz", ".br", ".mp4",
".webm", ".mp3", ".ogg", ".wasm", ".dmg", ".pdf", ".ico")
def safe_deploy(d):
out = {k: d.get(k) for k in DEPLOY_FIELDS}
if d.get("error_message"):
out["error_message"] = str(d["error_message"])[:500]
out["permalink"] = (d.get("links") or {}).get("permalink")
out["available_functions"] = [
{k: f.get(k) for k in ("n", "name", "d", "display_name", "a", "runtime", "g", "generator") if k in f}
for f in (d.get("available_functions") or []) if isinstance(f, dict)]
out["required_functions_count"] = len(d.get("required_functions") or [])
out["agent_runner_present"] = bool(d.get("agent_runner_id"))
out["database_branch_present"] = bool(d.get("database_branch_id"))
return out
def map_paths(remote_paths):
"""Map remote deploy paths to safe, collision free relative local paths. Returns (mapping, issues)."""
mapping, issues, seen_fold = {}, [], {}
files = set()
for rp in sorted(remote_paths):
parts = [p for p in rp.split("/") if p != ""]
bad = [p for p in parts if p in (".", "..")]
if bad or not parts:
issues.append({"remote_path": rp, "issue": "path traversal or empty component, renamed"})
parts = [("_dot_" if p in (".", "..") else p) for p in parts] or ["_empty_"]
fixed = []
for p in parts:
p = p.replace("\x00", "_")
if len(p.encode()) > 240:
h = hashlib.sha1(p.encode()).hexdigest()[:12]
p = p.encode()[:200].decode("utf-8", "ignore") + "~" + h
issues.append({"remote_path": rp, "issue": "component over 240 bytes, shortened"})
fixed.append(p)
local = "/".join(fixed)
key = unicodedata.normalize("NFC", local).casefold()
if key in seen_fold:
n = 2
while f"{key}~case{n}" in seen_fold:
n += 1
issues.append({"remote_path": rp, "issue": f"case or Unicode collision with {seen_fold[key]}"})
local = f"{local}~case{n}"
key = f"{key}~case{n}"
seen_fold[key] = rp
mapping[rp] = local
files.add(local)
# A path that is both a file and a parent directory of another file cannot exist on disk.
dirs = set()
for lp in files:
segs = lp.split("/")
for i in range(1, len(segs)):
dirs.add("/".join(segs[:i]))
for rp, lp in list(mapping.items()):
if lp in dirs:
mapping[rp] = lp + "~file"
issues.append({"remote_path": rp, "issue": "file also used as a directory by another path, renamed with ~file"})
return mapping, issues
class SiteRecovery:
def __init__(self, site_id, info):
self.site_id = site_id
self.info = info
self.dir = info["dir"]
self.meta = json.load(open(os.path.join(self.dir, "metadata", "site.json")))
self.name = self.meta["name"]
self.cas = os.path.join(self.dir, "originals", ".content_store")
def cas_path(self, sha):
return os.path.join(self.cas, sha[:2], sha)
def current_state(self):
st, s = api_json(f"/sites/{self.site_id}")
return (s.get("published_deploy") or {}).get("id") if s else None
def list_deploys(self):
deps, pages, st = paginate(f"/sites/{self.site_id}/deploys")
rec = {"retrieved_at": now(), "endpoint": "GET /api/v1/sites/{site_id}/deploys", "http_status": st,
"pages": pages, "count": len(deps), "deploys": [safe_deploy(d) for d in deps]}
write_json(os.path.join(self.dir, "metadata", "deploys.json"), rec)
return rec
def fetch(self, entry, deploy, current_id, current_by_sha):
"""Return (method, error). Writes verified bytes into the content store."""
sha, path = entry["sha"], entry["path"]
dest = self.cas_path(sha)
if os.path.exists(dest):
return "content_store_reuse", None
tries = []
if deploy["id"] == current_id:
tries.append(("documented_raw_current_deploy", path, True))
elif sha in current_by_sha:
tries.append(("documented_raw_same_sha_in_current_deploy", current_by_sha[sha], True))
tries.append(("deploy_permalink_sha1_verified", path, False))
errs = []
for method, p, raw in tries:
if raw:
url = f"{API}/sites/{self.site_id}/files/{urllib.parse.quote(p.lstrip('/'), safe='/')}"
st, h, body, _ = request(url, headers={"Content-Type": "application/vnd.bitballoon.v1.raw"})
else:
base = deploy.get("permalink") or deploy.get("deploy_ssl_url")
if not base:
errs.append("no permalink")
continue
url = base.rstrip("/") + urllib.parse.quote(p, safe="/")
st, h, body, _ = request(url, auth=False)
if st != 200:
errs.append(f"{method}: HTTP {st}")
continue
if sha1(body) != sha:
errs.append(f"{method}: sha1 mismatch (got {len(body)} bytes, manifest {entry.get('size')})")
continue
write_bytes_atomic(dest, body)
return method, None
return None, "; ".join(errs)
def recover_deploy(self, deploy, current_id, current_listing):
did = deploy["id"]
mdir = os.path.join(self.dir, "manifests")
files, pages, st = paginate(f"/deploys/{did}/files")
if st != 200:
return {"deploy_id": did, "status": "BLOCKED", "reason": f"deploy file listing HTTP {st}"}
foreign = sorted({f.get("deploy_id") for f in files} - {did})
write_json(os.path.join(mdir, f"deploy_{did}_files_listing.json"),
{"retrieved_at": now(), "endpoint": "GET /api/v1/deploys/{deploy_id}/files (verified read only)",
"pages": pages, "count": len(files), "foreign_deploy_ids_in_listing": foreign, "files": files})
current_by_sha = {f["sha"]: f["path"] for f in current_listing}
results = {}
need_bytes = sum(f.get("size") or 0 for f in files if not os.path.exists(self.cas_path(f["sha"])))
if free_gb() - need_bytes / 1e9 < MIN_FREE_GB:
return {"deploy_id": did, "status": "PENDING_DISK", "reason": f"needs {need_bytes/1e9:.2f} GB, free {free_gb():.1f} GB"}
started = now()
uniq = {}
for f in files:
uniq.setdefault(f["sha"], f)
with cf.ThreadPoolExecutor(max_workers=4) as ex:
futs = {ex.submit(self.fetch, f, deploy, current_id, current_by_sha): sha for sha, f in uniq.items()}
for i, fu in enumerate(cf.as_completed(futs), 1):
results[futs[fu]] = fu.result()
if i % 2000 == 0:
log(f" {self.name} {did}: {i}/{len(uniq)} unique contents")
finished = now()
mapping, issues = map_paths([f["path"] for f in files])
tree = os.path.join(self.dir, "originals", f"deploy_{did}")
entries, missing = [], []
for f in files:
method, err = results[f["sha"]]
local = mapping[f["path"]]
rec = {"remote_path": f["path"], "local_path": local, "sha1_manifest": f["sha"], "size_manifest": f.get("size"),
"mime_type": f.get("mime_type"), "method": method}
if err:
rec["error"] = err
missing.append(rec)
else:
dst = os.path.join(tree, local)
os.makedirs(os.path.dirname(dst), exist_ok=True)
if not os.path.exists(dst):
os.link(self.cas_path(f["sha"]), dst)
rec["size_local"] = os.path.getsize(dst)
rec["sha256"] = sha256_file(dst)
entries.append(rec)
status = "COMPLETE" if not missing and files else ("EMPTY_DEPLOY" if not files else ("PARTIAL" if len(missing) < len(files) else "FAILED"))
zip_info = self.make_zip(deploy, tree, entries) if status in ("COMPLETE", "PARTIAL") else None
manifest = {"deploy": deploy, "site_id": self.site_id, "site_name": self.name, "status": status,
"download_started": started, "download_finished": finished, "file_count_manifest": len(files),
"file_count_recovered": len(files) - len(missing), "missing_count": len(missing),
"bytes_manifest": sum(f.get("size") or 0 for f in files),
"bytes_recovered": sum(e.get("size_local") or 0 for e in entries),
"path_issues": issues, "zip": zip_info, "files": entries}
write_json(os.path.join(mdir, f"deploy_{did}_recovery.json"), manifest)
return {k: manifest[k] for k in ("status", "file_count_manifest", "file_count_recovered", "missing_count",
"bytes_recovered")} | {"deploy_id": did, "zip": zip_info}
def make_zip(self, deploy, tree, entries):
zname = f"{safe_name(self.name)}__{self.site_id}__DEPLOY__{deploy['id']}.zip"
zpath = os.path.join(ROOT, "zips", zname)
tmp = zpath + ".part"
try:
ts = datetime.datetime.fromisoformat((deploy.get("created_at") or "1980-01-01T00:00:00")[:19]).timetuple()[:6]
except ValueError:
ts = (1980, 1, 1, 0, 0, 0)
ts = max(ts, (1980, 1, 1, 0, 0, 0))
with zipfile.ZipFile(tmp, "w", allowZip64=True) as z:
for e in sorted(entries, key=lambda e: e["local_path"]):
if e.get("error"):
continue
zi = zipfile.ZipInfo(e["local_path"], date_time=ts)
zi.external_attr = 0o644 << 16
zi.compress_type = zipfile.ZIP_STORED if e["local_path"].lower().endswith(STORED_EXT) else zipfile.ZIP_DEFLATED
with open(os.path.join(tree, e["local_path"]), "rb") as src, z.open(zi, "w", force_zip64=True) as dst:
for chunk in iter(lambda: src.read(1 << 20), b""):
dst.write(chunk)
os.replace(tmp, zpath)
return {"path": os.path.relpath(zpath, ROOT), "bytes": os.path.getsize(zpath), "sha256": sha256_file(zpath)}
def run(self):
ck = checkpoint_read()
dep_rec = self.list_deploys()
current_id = self.current_state()
current_listing = []
p = os.path.join(self.dir, "manifests", "current_site_files_listing.json")
if current_id:
current_listing, _, _ = paginate(f"/sites/{self.site_id}/files")
ready = [d for d in dep_rec["deploys"] if d["state"] == "ready"]
ready.sort(key=lambda d: (d["id"] != current_id, d.get("created_at") or ""), reverse=False)
# Published first, then newest to oldest.
ready = [d for d in ready if d["id"] == current_id] + sorted([d for d in ready if d["id"] != current_id],
key=lambda d: d.get("created_at") or "", reverse=True)
summary = {"deploys_total": dep_rec["count"], "deploys_ready": len(ready),
"deploy_states": {}, "published_deploy_id": current_id}
for d in dep_rec["deploys"]:
summary["deploy_states"][d["state"]] = summary["deploy_states"].get(d["state"], 0) + 1
checkpoint_update(lambda c: c["sites"].setdefault(self.site_id, {}).update({"deploy_summary": summary}))
for d in ready:
prev = ck.get("deploys", {}).get(d["id"], {})
if prev.get("status") == "COMPLETE" and prev.get("zip") and os.path.exists(os.path.join(ROOT, prev["zip"]["path"])):
log(f"skip verified {self.name} {d['id']}")
continue
before = self.current_state()
res = self.recover_deploy(d, current_id, current_listing)
after = self.current_state()
res["published_deploy_before"], res["published_deploy_after"] = before, after
if d["id"] == current_id and not (before == after == current_id):
res["status"] = "INCONSISTENT_PUBLISHED_CHANGED"
res["site_id"] = self.site_id
res["site_name"] = self.name
checkpoint_update(lambda c: c["deploys"].__setitem__(d["id"], res))
log(f"{self.name} deploy {d['id']}: {res['status']} {res.get('file_count_recovered')}/{res.get('file_count_manifest')} files, "
f"{(res.get('bytes_recovered') or 0)/1e6:.1f} MB, free {free_gb():.0f} GB")
def main():
cls = json.load(open(os.path.join(ROOT, "checkpoints", "classification.json")))["sites"]
only = set(sys.argv[1:])
todo = sorted([(v["tier"], k, v) for k, v in cls.items() if v["include"] and (not only or v["name"] in only)])
log(f"phase 3 start: {len(todo)} included sites, free {free_gb():.0f} GB")
for tier, sid, info in todo:
try:
SiteRecovery(sid, info).run()
except Exception as e:
log(f"ERROR site {info['name']}: {type(e).__name__}: {e}")
checkpoint_update(lambda c: c["sites"].setdefault(sid, {}).update({"recovery_error": f"{type(e).__name__}: {e}"}))
log("phase 3 done")
if __name__ == "__main__":
main()