vosonandCursor
9c5aa5b610
Separate the local recognition, web publishing, and shared data paths while preserving direct script execution and existing site content. Co-authored-by: Cursor <cursoragent@cursor.com>
100 lines
3.4 KiB
Python
100 lines
3.4 KiB
Python
"""Fix climperor OSS public-read access (Block Public Access + policy)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import sys
|
|
|
|
import oss2
|
|
|
|
BUCKET = "climperor"
|
|
ENDPOINT = "https://oss-cn-shanghai.aliyuncs.com"
|
|
|
|
POLICY = {
|
|
"Version": "1",
|
|
"Statement": [
|
|
{
|
|
"Sid": "PublicReadAbilityVideos",
|
|
"Effect": "Allow",
|
|
"Principal": "*",
|
|
"Action": ["oss:GetObject", "oss:GetObjectAcl"],
|
|
"Resource": [f"acs:oss:*:*:{BUCKET}/ability-video/*"],
|
|
}
|
|
],
|
|
}
|
|
|
|
|
|
def main() -> None:
|
|
ak = os.environ["KEYZOO_ASSET_META_ACCESSKEY_ID"]
|
|
sk = os.environ["KEYZOO_ASSET_SECRET_ACCESSKEY_SECRET"]
|
|
auth = oss2.Auth(ak, sk)
|
|
b = oss2.Bucket(auth, ENDPOINT, BUCKET)
|
|
|
|
# 1) Try disable Block Public Access (required on newer Aliyun accounts).
|
|
try:
|
|
# oss2 >= 2.18: put_bucket_public_access_block(block_public_access=False)
|
|
if hasattr(b, "put_bucket_public_access_block"):
|
|
b.put_bucket_public_access_block(False)
|
|
print("public_access_block: disabled via SDK")
|
|
else:
|
|
# Raw REST: PUT /?publicAccessBlock with XML
|
|
xml = (
|
|
'<?xml version="1.0" encoding="UTF-8"?>'
|
|
"<PublicAccessBlockConfiguration>"
|
|
"<BlockPublicAccess>false</BlockPublicAccess>"
|
|
"</PublicAccessBlockConfiguration>"
|
|
)
|
|
resp = b._do("PUT", "", params={"publicAccessBlock": ""}, data=xml)
|
|
print(f"public_access_block: raw PUT status={resp.status}")
|
|
except Exception as exc: # noqa: BLE001
|
|
print(f"public_access_block_warn: {type(exc).__name__}: {exc}")
|
|
|
|
# 2) Bucket ACL public-read (may be denied by account policy).
|
|
try:
|
|
b.put_bucket_acl(oss2.BUCKET_ACL_PUBLIC_READ)
|
|
print("acl: public-read set")
|
|
except Exception as exc: # noqa: BLE001
|
|
print(f"acl_warn: {type(exc).__name__}: {exc}")
|
|
|
|
# 3) Bucket policy for anonymous GetObject under ability-video/.
|
|
try:
|
|
b.put_bucket_policy(json.dumps(POLICY))
|
|
print("policy: public GetObject on ability-video/*")
|
|
except Exception as exc: # noqa: BLE001
|
|
print(f"policy_fail: {type(exc).__name__}: {exc}")
|
|
raise SystemExit(1)
|
|
|
|
info = b.get_bucket_info()
|
|
print(f"acl_now: {info.acl.grant if info.acl else '?'}")
|
|
|
|
# 4) Probe one known object if any exist.
|
|
import urllib.request
|
|
|
|
sample = None
|
|
for obj in oss2.ObjectIterator(b, prefix="ability-video/", max_keys=1):
|
|
sample = obj.key
|
|
break
|
|
if not sample:
|
|
print("no objects yet to probe")
|
|
return
|
|
url = f"https://{BUCKET}.oss-cn-shanghai.aliyuncs.com/{sample}"
|
|
try:
|
|
req = urllib.request.Request(url, method="HEAD")
|
|
with urllib.request.urlopen(req, timeout=20) as resp:
|
|
print(f"probe: {resp.status} {resp.headers.get('Content-Type')} {url}")
|
|
except Exception as exc: # noqa: BLE001
|
|
print(f"probe_fail: {exc}")
|
|
# Try signed URL to confirm object exists
|
|
signed = b.sign_url("HEAD", sample, 60)
|
|
print(f"signed_head_url_len={len(signed)} (object exists check via SDK)")
|
|
try:
|
|
meta = b.head_object(sample)
|
|
print(f"sdk_head: ok content_type={meta.content_type} size={meta.content_length}")
|
|
except Exception as exc2: # noqa: BLE001
|
|
print(f"sdk_head_fail: {exc2}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|