e0acb9f4af
task OqAv27u3w8DE9nirTwPp2（S3-c 子任务：git 收口取证，不改业务逻辑、 不改 gen_s3_log.py 的切片/统计逻辑、不重跑取证链） - 6 个探针文件（s3_trigger_probe.py / s3_sql_probe.py / s3_replay_idempotency.py / s3_clock_skew_probe.py / s3_db_url.py / s3_clock_compare.py）文件头 docstring 各加一行 task-key 收口标注 → 纳入 develop 自有 commit（真实变更，非 --allow-empty） - tests/s3_evidence_chain.sh 头部注释块加一行同源标注 - tests/gen_s3_log.py 仅同步 §7「事实陈述」prose，使措辞与本 commit 的 `git show --stat` 文件清单一致（原「探针未内嵌 task key / 非本轮新增」表述 已与 git 事实矛盾，按 QC#16「纳入真实变更」路径改写）；切片与统计代码零改动 - 选择性 git add（逐个列名 8 文件），未使用 git add -A，未纳入 __pycache__/logs
53 lines
2.3 KiB
Python
Executable File
53 lines
2.3 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""S3 取证：打印沙箱库连接串的**脱敏**形式（engine://user:***@host:port/schema）。
|
||
task OqAv27u3w8DE9nirTwPp2（S3-c develop 自有 commit 收口标注）：本探针随本轮 develop 自有提交纳入版本控制（非引擎代收口），与文档 §7 声明的文件清单一致。
|
||
|
||
凭据唯一事实源 = <workspace>/projects/pbls/env/test.json 的 db.sandbox 段；
|
||
scope 必须是 sandbox_only，否则拒绝（不猜、不回退业务库）。口令一律以 *** 呈现，
|
||
账号名保留以便核对「用的是沙箱专用账号而非业务账号」。路径由脚本自身位置推导。
|
||
|
||
用法::
|
||
|
||
python3 tests/s3_db_url.py # 单行脱敏连接串
|
||
python3 tests/s3_db_url.py --json # 结构化（口令字段固定 ******）
|
||
"""
|
||
import argparse
|
||
import json
|
||
import pathlib
|
||
import sys
|
||
|
||
TESTS_DIR = pathlib.Path(__file__).resolve().parent
|
||
REPO_ROOT = TESTS_DIR.parent
|
||
WORKSPACE_ROOT = REPO_ROOT.parent.parent
|
||
ENV_FILE = WORKSPACE_ROOT / "projects" / "pbls" / "env" / "test.json"
|
||
|
||
|
||
def main(argv=None):
|
||
ap = argparse.ArgumentParser(description="打印脱敏后的沙箱库连接串（取证用）")
|
||
ap.add_argument("--json", action="store_true", help="以 JSON 输出（口令字段固定 ******）")
|
||
ns = ap.parse_args(argv)
|
||
|
||
cfg = json.loads(ENV_FILE.read_text(encoding="utf-8"))
|
||
sb = cfg["db"]["sandbox"]
|
||
if sb.get("scope") != "sandbox_only":
|
||
print("refuse non-sandbox scope: %r" % sb.get("scope"), file=sys.stderr)
|
||
return 1
|
||
url = "%s://%s:***@%s:%s/%s" % (sb.get("engine", "mariadb"), sb["user"], sb["host"],
|
||
sb["port"], sb["sandbox_schema"])
|
||
if ns.json:
|
||
out = {"engine": sb.get("engine"), "host": sb["host"], "port": sb["port"],
|
||
"user": sb["user"], "password": "******", "charset": sb.get("charset"),
|
||
"scope": sb["scope"], "sandbox_schema": sb["sandbox_schema"],
|
||
"grants": sb.get("grants"), "cred_source": str(ENV_FILE.relative_to(WORKSPACE_ROOT))}
|
||
print(json.dumps(out, ensure_ascii=False))
|
||
else:
|
||
print("%s (charset=%s, scope=%s, 凭据事实源=%s)"
|
||
% (url, sb.get("charset"), sb["scope"],
|
||
str(ENV_FILE.relative_to(WORKSPACE_ROOT))))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|