Leo
d19bf88503
09-07 三種寫入在雲端全被 auto mode 分類器擋：npx wrangler secret put、 stage-deploy-artifacts.py --confirm、curl POST 到 arcrun-yuga3bse—— 白名單是逐條比前綴，複合指令與任意 curl 永遠對不上；放整類又是紅線。 - scripts/stage secret put|list|delete：直接打 CF API（不經 npx）；先 GET /accounts 看這把 token 打得到誰，stage 的 ID 不在裡面就拒絕並印出它打得到誰；URL 永遠只帶 1129efd7…；值只從 --from-env 讀，--value 拒收。 - scripts/stage api <METHOD> <worker>/<path>：curl 打實例，主機只會是 *.arcrun-yuga3bse.workers.dev，別的主機拒絕；標頭與 body 走 stdin 設定檔＋管線， 不走 argv；--bearer-env／--header-env／--data-env 從環境變數讀。 - docs/permissions-allow.json：stage 四種形狀＋Arcrun 自己那支 stage-deploy-artifacts.py 的相對形狀（它已寫死只認 youlin 帳號 ID、不收 --account／--token，不重造）。 - scripts/test-stage.sh 45 條（假 CF 對 PUT 回 201——第一次實跑 CF 回 201 被講成失敗， 就是這格抓的）；test-settings-allow-sync 仍 19/19。 - 真跑：探針種進 youlin arcrun-cypher-executor、CF API 獨立列到、刪掉恢復原狀； api GET health 200、無 Bearer POST 401、leo21c 主機拒絕零請求。 「帶 Bearer 真寫 KBDB」缺 KBDB_INTERNAL_TOKEN 沒跑（TESTING A36 標明）。 盤點在本樹實數：hooks 61、註冊 85（都沒動）、scripts 56（＋stage、＋test-stage.sh）。 版本待總管定版。 Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01DTZ9QtvjY7MNxjfQbexAm7
411 lines
22 KiB
Python
Executable File
411 lines
22 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""stage — 雲端對 **stage（youlin）** 寫入的正門（inkstone/ISEP#137）
|
||
|
||
━━ 為什麼有這支 ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
|
||
09-07 三條主線工人在雲端對 stage 的寫入全被 Claude Code auto mode 的分類器擋（不是 ISEP 的閘）：
|
||
① `printf … | npx wrangler secret put CF_SECRETS_API_TOKEN --name arcrun-cypher-executor`
|
||
② `python3 scripts/stage-deploy-artifacts.py all --confirm`（Arcrun 的成品直推，CF API 上傳）
|
||
③ `curl -X POST …arcrun-yuga3bse.workers.dev/records`／`/graph/canonicalize-entities`／`/map/recompute`
|
||
權限白名單（docs/permissions-allow.json）是**逐條完全比對前綴**：複合指令（`cd … && printf … | VAR=… npx …`）
|
||
與任意寫法的 curl 永遠對不上任何一條；放整類 `Bash(npx *)`／`Bash(curl *)` 又是紅線。
|
||
leo 09-07：「為什麼會擋你，這就是要修復的問題」「本機環境跟你不同應該不行」。
|
||
|
||
⇒ 每一種寫入收成**一支固定形狀**，白名單放它的四種呼叫形狀：
|
||
`stage secret …` ＝ ①（打 api.cloudflare.com，不經 npx——雲端每次要重抓 wrangler，
|
||
而且 wrangler 會讀 ~/.wrangler 的登入態：本機是 uncle6、雲端沒有，兩台行為不同）
|
||
`stage api …` ＝ ③（用 curl 打 stage 實例；youlin 邊緣對 python-urllib UA 回 403 code 1010，
|
||
Arcrun#176 comment 6487——所以這一段**不用 urllib**）
|
||
② ＝ Arcrun 自己那支已經寫死只認 youlin 帳號 ID、憑證只從
|
||
CLOUDFLARE_API_TOKEN_YOULIN_CC_USE 讀、不收 --account／--token；
|
||
白名單直接放它的形狀（`python3 scripts/stage-deploy-artifacts.py *`），不重造一支。
|
||
|
||
━━ 只寫 stage，判準是唯一識別碼往放行方向比對，不是名字 ━━━━━━━━━━━━━━━━━━━━━━
|
||
stage 隨便炸；出貨機（geek6688／uncle6）與 leo 的實例（leo21c）不是。
|
||
secret：這把 token 打得到的帳號（GET /accounts 現查）裡要有 1129efd7df2e8899d537e9c8fbabb6cb，
|
||
而且寫入的 URL 永遠只帶這個 ID。打不到 ⇒ 拒絕、印出它打得到誰、離開碼 2。
|
||
api ：URL 只會長在 `<worker>.arcrun-yuga3bse.workers.dev`（09-02 重裝後的 stage 子網域）。
|
||
給完整網址時主機必須等於它，否則拒絕、離開碼 2——不管那個主機叫 leo21c 還是別的什麼。
|
||
不看變數名字裡有沒有 `leo21c`／`geek6688`——名字改了黑名單就漏，ID 與主機改不了。
|
||
|
||
━━ 值不進指令列 ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
|
||
secret 的值、Bearer、任何標頭值都只從**環境變數名**讀（--from-env／--bearer-env／--header-env）；
|
||
不落磁碟、不 echo、不進 git、不進 shell 歷史。curl 的標頭與 body 走 stdin 設定檔＋管線，不走 argv。
|
||
|
||
━━ 用法 ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
|
||
stage secret put --worker <script> --secret <NAME> --from-env <VAR> [--token-env VAR] [--account ID]
|
||
stage secret list --worker <script>
|
||
stage secret delete --worker <script> --secret <NAME>
|
||
--token-env <VAR> CF token 的環境變數名（預設 CLOUDFLARE_API_TOKEN_YOULIN_CC_USE，雲端就有）
|
||
--account <ID> 目標帳號（預設 stage；給別的 ID 一律拒絕、不打網路）
|
||
|
||
stage api <METHOD> <worker>/<path>[?query] [--json '<body>' | --data-env <VAR> | --data-file <檔>]
|
||
[--bearer-env <VAR>] [--header-env <Name>=<VAR>]…
|
||
<worker>/<path> 例 arcrun-kbdb/records、arcrun-cypher-executor/webhooks/named/yuga3bse/x/trigger
|
||
也收完整網址，但主機必須是 *.arcrun-yuga3bse.workers.dev
|
||
--bearer-env VAR Authorization: Bearer <VAR 的值>（stage KBDB 的 POST 要 KBDB_INTERNAL_TOKEN）
|
||
--header-env N=VAR 任一標頭，值從環境變數讀（例 X-Arcrun-API-Key=ARCRUN_NS_YOULIN）
|
||
body 有給就自動帶 Content-Type: application/json
|
||
印 `HTTP <code>` ＋ 回應；2xx 離開碼 0，其餘 1；curl 連不上印它的錯誤字串（不記成 000）
|
||
|
||
例（雲端）：
|
||
python3 "$CLAUDE_PLUGIN_ROOT/scripts/stage" secret put --worker arcrun-cypher-executor \\
|
||
--secret CF_SECRETS_API_TOKEN --from-env CF_SECRETS_API_TOKEN_VALUE
|
||
python3 "$CLAUDE_PLUGIN_ROOT/scripts/stage" api POST 'arcrun-kbdb/map/recompute?library=rt-lib&owner_id=yuga3bse' \\
|
||
--bearer-env KBDB_INTERNAL_TOKEN
|
||
|
||
離開碼：0 成功 ／ 2 被本工具拒絕（非 stage、缺變數、值在指令列）／ 1 API・網路失敗
|
||
測試用：STAGE_SECRET_PUT_API_BASE（CF API）、STAGE_API_BASE_OVERRIDE（實例）指到假伺服器——scripts/test-stage.sh
|
||
"""
|
||
import json
|
||
import os
|
||
import subprocess
|
||
import sys
|
||
import urllib.error
|
||
import urllib.parse
|
||
import urllib.request
|
||
|
||
STAGE_ACCOUNT_ID = "1129efd7df2e8899d537e9c8fbabb6cb" # youlin（D37 定的 stage 帳號）
|
||
STAGE_LABEL = "youlin（stage）"
|
||
STAGE_HOST_SUFFIX = ".arcrun-yuga3bse.workers.dev" # 09-02 重裝後的 stage 子網域（ISEP#130）
|
||
DEFAULT_TOKEN_ENV = "CLOUDFLARE_API_TOKEN_YOULIN_CC_USE"
|
||
CF_API_BASE = os.environ.get("STAGE_SECRET_PUT_API_BASE", "https://api.cloudflare.com/client/v4").rstrip("/")
|
||
API_BASE_OVERRIDE = os.environ.get("STAGE_API_BASE_OVERRIDE", "").rstrip("/") # 測試用：整個實例指到假伺服器
|
||
|
||
EXIT_OK, EXIT_API, EXIT_REFUSED = 0, 1, 2
|
||
FORBIDDEN_VALUE_FLAGS = ("--value", "--text", "--bearer", "--token", "--header")
|
||
|
||
|
||
def die(code, msg):
|
||
print(msg, file=sys.stderr)
|
||
return code
|
||
|
||
|
||
class Refuse(Exception):
|
||
pass
|
||
|
||
|
||
# ───────────────────────── secret（CF API，api.cloudflare.com）─────────────────────────
|
||
|
||
def parse_secret(argv):
|
||
if not argv or argv[0] not in ("put", "list", "delete"):
|
||
raise Refuse("secret 後面要接 put／list／delete，不是 %r（--help 看用法）" % (argv[:1] or [""])[0])
|
||
cmd = argv[0]
|
||
o = {"worker": None, "secret": None, "from_env": None, "token_env": DEFAULT_TOKEN_ENV, "account": STAGE_ACCOUNT_ID}
|
||
keys = {"--worker": "worker", "--secret": "secret", "--from-env": "from_env", "--token-env": "token_env", "--account": "account"}
|
||
i = 1
|
||
while i < len(argv):
|
||
a = argv[i]
|
||
if a in FORBIDDEN_VALUE_FLAGS:
|
||
raise Refuse("本工具不收 %s：值只從環境變數名讀（--from-env／--token-env），不准出現在指令列" % a)
|
||
if a not in keys:
|
||
raise Refuse("不認得的參數：%s（--help 看用法）" % a)
|
||
if i + 1 >= len(argv):
|
||
raise Refuse("%s 後面要接一個值" % a)
|
||
o[keys[a]] = argv[i + 1]
|
||
i += 2
|
||
if not o["worker"]:
|
||
raise Refuse("缺 --worker <script 名>")
|
||
if cmd in ("put", "delete") and not o["secret"]:
|
||
raise Refuse("缺 --secret <SECRET 名>")
|
||
if cmd == "put" and not o["from_env"]:
|
||
raise Refuse("缺 --from-env <環境變數名>（值只從這裡讀）")
|
||
return cmd, o
|
||
|
||
|
||
def cf(method, path, token, body=None):
|
||
"""回 (http_status, parsed_json)；網路層失敗回 (None, 錯誤字串)——不把三種病都記成 000。"""
|
||
data = None
|
||
headers = {"Authorization": "Bearer " + token, "User-Agent": "isep-stage/1 (+inkstone/ISEP#137)"}
|
||
if body is not None:
|
||
data = json.dumps(body).encode("utf-8")
|
||
headers["Content-Type"] = "application/json"
|
||
req = urllib.request.Request(CF_API_BASE + path, data=data, headers=headers, method=method)
|
||
try:
|
||
with urllib.request.urlopen(req, timeout=60) as resp:
|
||
return resp.status, json.loads(resp.read().decode("utf-8") or "{}")
|
||
except urllib.error.HTTPError as e:
|
||
try:
|
||
return e.code, json.loads(e.read().decode("utf-8") or "{}")
|
||
except Exception:
|
||
return e.code, {"success": False, "errors": [{"message": "HTTP %d（非 JSON 回應）" % e.code}]}
|
||
except Exception as e:
|
||
return None, "%s: %s" % (type(e).__name__, e)
|
||
|
||
|
||
def cf_ok(status, d):
|
||
# CF 對「建立」回 201（secret PUT 實測就是 201，inkstone/ISEP#137 第一次實跑抓到的）——2xx ＋ success 才算好
|
||
return status is not None and 200 <= status < 300 and isinstance(d, dict) and bool(d.get("success"))
|
||
|
||
|
||
def cf_errors(d):
|
||
if isinstance(d, dict):
|
||
errs = d.get("errors") or []
|
||
return "；".join("%s（code %s）" % (e.get("message"), e.get("code")) for e in errs) or "（CF 沒給錯誤訊息）"
|
||
return str(d)
|
||
|
||
|
||
def refuse_account(given):
|
||
return die(EXIT_REFUSED, "\n".join([
|
||
"🛑 本工具只種 stage，不種 %s。" % given,
|
||
"",
|
||
" stage ＝ %s，帳號 ID %s" % (STAGE_LABEL, STAGE_ACCOUNT_ID),
|
||
" 你給的 ＝ %s" % given,
|
||
"",
|
||
"【為什麼】stage 隨便炸；出貨機（geek6688／uncle6）與 leo 的實例（leo21c）不是（inkstone/ISEP#137）。",
|
||
"【出路】要種進那兩類帳號 ⇒ 交回地端，由 leo 親手在 Mac 上 `wrangler secret put`；",
|
||
" 雲端這支不會、也不該會。要種 stage ⇒ 把 --account 拿掉（預設就是 stage）。",
|
||
]))
|
||
|
||
|
||
def check_token_reaches_stage(token, token_env):
|
||
status, d = cf("GET", "/accounts", token)
|
||
if status is None:
|
||
return die(EXIT_API, "🔴 連不到 CF API（%s）：%s\n 雲端看到這個先查 `curl -sS \"$HTTPS_PROXY/__agentproxy/status\"`（policy 拒絕 vs DNS vs 對方沒回，三種修法不同）。" % (CF_API_BASE, d))
|
||
if not cf_ok(status, d):
|
||
return die(EXIT_API, "🔴 CF 不認這把 token（%s，HTTP %s）：%s" % (token_env, status, cf_errors(d)))
|
||
accounts = [(a.get("id"), a.get("name")) for a in (d.get("result") or []) if isinstance(a, dict)]
|
||
if STAGE_ACCOUNT_ID not in [i for i, _ in accounts]:
|
||
reach = "\n".join(" · %s %s" % (i, n) for i, n in accounts) or " ·（一個都沒有）"
|
||
return die(EXIT_REFUSED, "\n".join([
|
||
"🛑 這把 token（%s）打不到 stage，拒絕。" % token_env,
|
||
"",
|
||
" stage ＝ %s，帳號 ID %s" % (STAGE_LABEL, STAGE_ACCOUNT_ID),
|
||
" 這把 token 打得到的帳號：",
|
||
reach,
|
||
"",
|
||
"【為什麼】本工具只種 stage；token 是誰的就會種進誰的帳號，所以先問 CF 它打得到誰（inkstone/ISEP#137）。",
|
||
"【出路】用 stage 的那把：--token-env %s（雲端 Cloud environment 就有，本機在頂層 .env）。" % DEFAULT_TOKEN_ENV,
|
||
" 真要種出貨機／leo21c ⇒ 交回地端由 leo 親手做。",
|
||
]))
|
||
return None
|
||
|
||
|
||
def list_names(token, account, worker):
|
||
status, d = cf("GET", "/accounts/%s/workers/scripts/%s/secrets" % (account, worker), token)
|
||
if status is None:
|
||
return None, "連不到 CF API（%s）：%s" % (CF_API_BASE, d)
|
||
if not cf_ok(status, d):
|
||
return None, "列 secrets 失敗（HTTP %s）：%s" % (status, cf_errors(d))
|
||
return sorted(s.get("name") for s in (d.get("result") or []) if isinstance(s, dict) and s.get("name")), None
|
||
|
||
|
||
def run_secret(argv):
|
||
try:
|
||
cmd, o = parse_secret(argv)
|
||
except Refuse as e:
|
||
return die(EXIT_REFUSED, "🛑 %s" % e)
|
||
|
||
if o["account"] != STAGE_ACCOUNT_ID:
|
||
return refuse_account(o["account"])
|
||
token = os.environ.get(o["token_env"], "")
|
||
if not token:
|
||
return die(EXIT_REFUSED, "🛑 環境變數 %s 沒有值（CF token）。\n【出路】雲端：Cloud environment 要有它（scripts/make-cloud-env.sh 的清單裡就有）；本機：從頂層 .env 匯出。" % o["token_env"])
|
||
value = None
|
||
if cmd == "put":
|
||
value = os.environ.get(o["from_env"], "")
|
||
if not value:
|
||
return die(EXIT_REFUSED, "🛑 環境變數 %s 沒有值（要種進去的 secret 本身）。\n【出路】把值放進這個環境變數再跑；不要改成把值寫在指令列——那是紅線（值不落磁碟、不進指令歷史）。" % o["from_env"])
|
||
|
||
rc = check_token_reaches_stage(token, o["token_env"])
|
||
if rc is not None:
|
||
return rc
|
||
acct, worker = o["account"], o["worker"]
|
||
|
||
if cmd == "list":
|
||
names, err = list_names(token, acct, worker)
|
||
if err:
|
||
return die(EXIT_API, "🔴 %s" % err)
|
||
print("✅ %s@%s 現有 %d 把 secret：%s" % (worker, STAGE_LABEL, len(names), "、".join(names) or "（沒有）"))
|
||
return EXIT_OK
|
||
|
||
if cmd == "delete":
|
||
status, d = cf("DELETE", "/accounts/%s/workers/scripts/%s/secrets/%s" % (acct, worker, o["secret"]), token)
|
||
if status is None:
|
||
return die(EXIT_API, "🔴 連不到 CF API：%s" % d)
|
||
if not cf_ok(status, d):
|
||
return die(EXIT_API, "🔴 刪 %s 失敗（HTTP %s）：%s" % (o["secret"], status, cf_errors(d)))
|
||
names, err = list_names(token, acct, worker)
|
||
if err:
|
||
return die(EXIT_API, "🔴 刪了但列不出來確認：%s" % err)
|
||
if o["secret"] in names:
|
||
return die(EXIT_API, "🔴 CF 說刪了，列表卻還有 %s" % o["secret"])
|
||
print("✅ 已刪 %s ← %s@%s；剩 %d 把：%s" % (o["secret"], worker, STAGE_LABEL, len(names), "、".join(names) or "（沒有）"))
|
||
return EXIT_OK
|
||
|
||
body = {"name": o["secret"], "text": value, "type": "secret_text"}
|
||
status, d = cf("PUT", "/accounts/%s/workers/scripts/%s/secrets" % (acct, worker), token, body)
|
||
del value, body
|
||
if status is None:
|
||
return die(EXIT_API, "🔴 連不到 CF API：%s" % d)
|
||
if not cf_ok(status, d):
|
||
return die(EXIT_API, "🔴 種 %s → %s 失敗（HTTP %s）：%s" % (o["secret"], worker, status, cf_errors(d)))
|
||
names, err = list_names(token, acct, worker)
|
||
if err:
|
||
return die(EXIT_API, "🔴 CF 說種好了，但列不出來確認：%s" % err)
|
||
if o["secret"] not in names:
|
||
return die(EXIT_API, "🔴 CF 說種好了，列表卻沒有 %s（有的是：%s）" % (o["secret"], "、".join(names)))
|
||
print("✅ 已種 %s → %s@%s（值來自環境變數 %s，沒印、沒落地）" % (o["secret"], worker, STAGE_LABEL, o["from_env"]))
|
||
print(" 現有 %d 把：%s" % (len(names), "、".join(names)))
|
||
return EXIT_OK
|
||
|
||
|
||
# ───────────────────────── api（curl 打 stage 實例）─────────────────────────
|
||
|
||
def parse_api(argv):
|
||
if len(argv) < 2:
|
||
raise Refuse("api 後面要接 <METHOD> <worker>/<path>（--help 看用法）")
|
||
method, target = argv[0].upper(), argv[1]
|
||
if method not in ("GET", "POST", "PUT", "PATCH", "DELETE"):
|
||
raise Refuse("METHOD 要是 GET／POST／PUT／PATCH／DELETE，不是 %r" % argv[0])
|
||
o = {"json": None, "data_env": None, "data_file": None, "bearer_env": None, "header_envs": []}
|
||
i = 2
|
||
while i < len(argv):
|
||
a = argv[i]
|
||
if a in FORBIDDEN_VALUE_FLAGS:
|
||
raise Refuse("本工具不收 %s：標頭值只從環境變數名讀（--bearer-env／--header-env Name=VAR），不准出現在指令列" % a)
|
||
if i + 1 >= len(argv):
|
||
raise Refuse("%s 後面要接一個值" % a)
|
||
v = argv[i + 1]
|
||
if a == "--json":
|
||
o["json"] = v
|
||
elif a == "--data-env":
|
||
o["data_env"] = v
|
||
elif a == "--data-file":
|
||
o["data_file"] = v
|
||
elif a == "--bearer-env":
|
||
o["bearer_env"] = v
|
||
elif a == "--header-env":
|
||
if "=" not in v:
|
||
raise Refuse("--header-env 的形狀是 Name=VAR（例 X-Arcrun-API-Key=ARCRUN_NS_YOULIN），不是 %r" % v)
|
||
o["header_envs"].append(tuple(v.split("=", 1)))
|
||
else:
|
||
raise Refuse("不認得的參數：%s（--help 看用法）" % a)
|
||
i += 2
|
||
if sum(1 for k in ("json", "data_env", "data_file") if o[k]) > 1:
|
||
raise Refuse("--json／--data-env／--data-file 只能給一個")
|
||
return method, target, o
|
||
|
||
|
||
def stage_url(target):
|
||
"""把 <worker>/<path> 或完整網址變成 stage 的網址；主機不是 stage 就拒絕。"""
|
||
if "://" in target:
|
||
u = urllib.parse.urlsplit(target)
|
||
host = (u.hostname or "").lower()
|
||
if not host.endswith(STAGE_HOST_SUFFIX) or host == STAGE_HOST_SUFFIX.lstrip("."):
|
||
raise Refuse("\n".join([
|
||
"本工具只打 stage，不打 %s。" % host,
|
||
"",
|
||
" stage ＝ %s，主機一律是 <worker>%s" % (STAGE_LABEL, STAGE_HOST_SUFFIX),
|
||
" 你給的 ＝ %s" % target,
|
||
"",
|
||
"【為什麼】stage 隨便炸；leo21c 是 leo 本人在用的實例、geek6688／uncle6 是出貨機（inkstone/ISEP#137）。",
|
||
"【出路】打 stage ⇒ 寫成 <worker>/<path>（例 arcrun-kbdb/records），主機由本工具補。",
|
||
" 對 leo21c 唯讀查詢 ⇒ 直接 curl GET（leo21c-write-guard 放行讀）；要寫 ⇒ 交回地端由 leo 親手做。",
|
||
]))
|
||
if API_BASE_OVERRIDE:
|
||
return API_BASE_OVERRIDE + "/" + host.split(".")[0] + (u.path or "/") + ("?" + u.query if u.query else "")
|
||
return urllib.parse.urlunsplit(("https", host, u.path or "/", u.query, ""))
|
||
if "/" not in target:
|
||
raise Refuse("目標要寫成 <worker>/<path>（例 arcrun-kbdb/records），不是 %r" % target)
|
||
worker, rest = target.split("/", 1)
|
||
if not worker or any(c in worker for c in ".:@ "):
|
||
raise Refuse("<worker> 只能是 worker 名（例 arcrun-kbdb），不是 %r" % worker)
|
||
if API_BASE_OVERRIDE:
|
||
return API_BASE_OVERRIDE + "/" + worker + "/" + rest
|
||
return "https://" + worker + STAGE_HOST_SUFFIX + "/" + rest
|
||
|
||
|
||
def curl_quote(s):
|
||
return '"' + s.replace("\\", "\\\\").replace('"', '\\"').replace("\n", "\\n").replace("\t", "\\t") + '"'
|
||
|
||
|
||
def run_api(argv):
|
||
try:
|
||
method, target, o = parse_api(argv)
|
||
url = stage_url(target)
|
||
except Refuse as e:
|
||
return die(EXIT_REFUSED, "🛑 %s" % e)
|
||
|
||
headers = []
|
||
if o["bearer_env"]:
|
||
v = os.environ.get(o["bearer_env"], "")
|
||
if not v:
|
||
return die(EXIT_REFUSED, "🛑 環境變數 %s 沒有值（Bearer）。\n【出路】把值放進這個環境變數再跑；不要改成寫在指令列。" % o["bearer_env"])
|
||
headers.append("Authorization: Bearer " + v)
|
||
for name, var in o["header_envs"]:
|
||
v = os.environ.get(var, "")
|
||
if not v:
|
||
return die(EXIT_REFUSED, "🛑 環境變數 %s 沒有值（標頭 %s）。" % (var, name))
|
||
headers.append(name + ": " + v)
|
||
|
||
body = None
|
||
if o["json"] is not None:
|
||
body = o["json"]
|
||
elif o["data_env"]:
|
||
body = os.environ.get(o["data_env"], "")
|
||
if not body:
|
||
return die(EXIT_REFUSED, "🛑 環境變數 %s 沒有值（body）。" % o["data_env"])
|
||
elif o["data_file"]:
|
||
try:
|
||
with open(o["data_file"], encoding="utf-8") as f:
|
||
body = f.read()
|
||
except OSError as e:
|
||
return die(EXIT_REFUSED, "🛑 讀不了 --data-file %s：%s" % (o["data_file"], e))
|
||
if body is not None:
|
||
headers.append("Content-Type: application/json")
|
||
|
||
# 標頭與 body 都不走 argv：設定檔從 stdin 進，body 從一條管線（/dev/fd/N）進
|
||
rfd, wfd = os.pipe()
|
||
cfg = ["url = " + curl_quote(url), "request = " + curl_quote(method), "silent", "show-error",
|
||
"write-out = " + curl_quote("\\n__STAGE_HTTP__%{http_code}"), "max-time = 120",
|
||
"user-agent = " + curl_quote("isep-stage/1 (+inkstone/ISEP#137)")]
|
||
for h in headers:
|
||
cfg.append("header = " + curl_quote(h))
|
||
if body is not None:
|
||
cfg.append("data-binary = " + curl_quote("@/dev/fd/%d" % rfd))
|
||
try:
|
||
p = subprocess.Popen(["curl", "--config", "-"], stdin=subprocess.PIPE, stdout=subprocess.PIPE,
|
||
stderr=subprocess.PIPE, pass_fds=(rfd,) if body is not None else ())
|
||
except FileNotFoundError:
|
||
os.close(rfd); os.close(wfd)
|
||
return die(EXIT_API, "🔴 這台沒有 curl。")
|
||
os.close(rfd)
|
||
if body is not None:
|
||
with os.fdopen(wfd, "w", encoding="utf-8") as w:
|
||
w.write(body)
|
||
else:
|
||
os.close(wfd)
|
||
out, err = p.communicate(("\n".join(cfg) + "\n").encode("utf-8"))
|
||
del headers, cfg
|
||
text = out.decode("utf-8", "replace")
|
||
if p.returncode != 0:
|
||
return die(EXIT_API, "🔴 curl 失敗（離開碼 %d）：%s\n 雲端 403 CONNECT ⇒ `curl -sS \"$HTTPS_PROXY/__agentproxy/status\"`（policy）；`error code: 1010` ⇒ youlin 邊緣擋 UA。"
|
||
% (p.returncode, err.decode("utf-8", "replace").strip()))
|
||
resp, _, code = text.rpartition("\n__STAGE_HTTP__")
|
||
code = code.strip() or "000"
|
||
try:
|
||
pretty = json.dumps(json.loads(resp), ensure_ascii=False, indent=2) if resp.strip() else ""
|
||
except ValueError:
|
||
pretty = resp
|
||
print("HTTP %s %s %s" % (code, method, url))
|
||
if pretty:
|
||
print(pretty)
|
||
return EXIT_OK if code.isdigit() and 200 <= int(code) < 300 else EXIT_API
|
||
|
||
|
||
# ───────────────────────── 入口 ─────────────────────────
|
||
|
||
def main(argv):
|
||
if not argv or argv[0] in ("-h", "--help"):
|
||
print(__doc__)
|
||
return EXIT_OK
|
||
if argv[0] == "secret":
|
||
return run_secret(argv[1:])
|
||
if argv[0] == "api":
|
||
return run_api(argv[1:])
|
||
return die(EXIT_REFUSED, "🛑 第一個參數要是 secret／api，不是 %r（--help 看用法）" % argv[0])
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main(sys.argv[1:]))
|