|
|
@@ -0,0 +1,124 @@
|
|
|
+# -*- coding: utf-8 -*-
|
|
|
+"""全库体检：机构密钥有没有出现在**任何 git 跟踪文件**里（不限 logs/）。
|
|
|
+
|
|
|
+为什么要有它：_lsy-key-scan.py 只扫 logs/ 下的 _lsy* 产物 —— 那是「我准备提交什么」的量具。
|
|
|
+但「仓库里有没有这个值」是另一个问题，射程不同：本探针扫 git ls-files 的**全部 1233 个**跟踪
|
|
|
+文件的工作区版本，外加 HEAD:application.yml（索引/提交里那一份）。
|
|
|
+两个量具都留着：前者窄而针对提交动作，后者宽而针对仓库状态。
|
|
|
+
|
|
|
+只打：密钥长度、密钥 sha256 前 16 位（不可逆指纹）、文件名、命中次数、**命中行的字段名**
|
|
|
+（该行里的密钥整段替换成 <KEY>）。绝不打印密钥本身或其任何子串。
|
|
|
+
|
|
|
+正对照：工作区的 application.yml 必须命中 ≥1 次，否则自判「探针失效」并以非零退出。
|
|
|
+
|
|
|
+用法：
|
|
|
+ python logs/_lsy-repo-keyscan.py # 计数版（默认，打在 stdout）
|
|
|
+ python logs/_lsy-repo-keyscan.py --ctx # 定位版：每个命中处的整行（密钥已替换）
|
|
|
+"""
|
|
|
+import io
|
|
|
+import os
|
|
|
+import re
|
|
|
+import subprocess
|
|
|
+import sys
|
|
|
+from hashlib import sha256
|
|
|
+
|
|
|
+YML = "java/src/main/resources/application.yml"
|
|
|
+VAR = re.compile(r"^\$\{([A-Za-z0-9_]+):(.*)\}$")
|
|
|
+
|
|
|
+
|
|
|
+def read_block(path, top):
|
|
|
+ lines = io.open(path, encoding="utf-8", errors="replace").read().split("\n")
|
|
|
+ start = None
|
|
|
+ for i, ln in enumerate(lines):
|
|
|
+ if re.match(r"^%s:\s*$" % re.escape(top), ln):
|
|
|
+ start = i
|
|
|
+ break
|
|
|
+ if start is None:
|
|
|
+ raise SystemExit("找不到顶层段 %s:" % top)
|
|
|
+ out = {}
|
|
|
+ for j in range(start + 1, len(lines)):
|
|
|
+ ln = lines[j]
|
|
|
+ if ln.strip() == "" or ln.lstrip().startswith("#"):
|
|
|
+ continue
|
|
|
+ if not ln.startswith((" ", "\t")):
|
|
|
+ break
|
|
|
+ m = re.match(r"^\s+([A-Za-z0-9_.\-]+):\s*(.*)$", ln)
|
|
|
+ if m:
|
|
|
+ out[m.group(1)] = re.split(r"\s+#", m.group(2), maxsplit=1)[0].strip().strip("'\"")
|
|
|
+ return out
|
|
|
+
|
|
|
+
|
|
|
+cfg = read_block(YML, "leshouying")
|
|
|
+raw = cfg.get("key", "")
|
|
|
+m = VAR.match(raw)
|
|
|
+KEY = (os.environ.get(m.group(1)) or m.group(2)) if m else raw
|
|
|
+needle = KEY.encode("utf-8")
|
|
|
+print("密钥长度 = %d, sha256[:16] = %s（只打长度与不可逆指纹）"
|
|
|
+ % (len(KEY), sha256(needle).hexdigest()[:16]))
|
|
|
+
|
|
|
+pos = io.open(YML, "rb").read().count(needle)
|
|
|
+print("正对照 工作区 application.yml 命中 = %d %s"
|
|
|
+ % (pos, "OK" if pos >= 1 else "!! 探针失效，体检作废"))
|
|
|
+if pos < 1:
|
|
|
+ raise SystemExit(1)
|
|
|
+
|
|
|
+tracked = subprocess.run(["git", "ls-files", "-z"], capture_output=True).stdout.split(b"\0")
|
|
|
+tracked = [t.decode("utf-8", "replace") for t in tracked if t]
|
|
|
+
|
|
|
+bad, missing = [], 0
|
|
|
+for f in tracked:
|
|
|
+ try:
|
|
|
+ c = io.open(f, "rb").read().count(needle)
|
|
|
+ except Exception:
|
|
|
+ missing += 1
|
|
|
+ continue
|
|
|
+ if c:
|
|
|
+ bad.append((f, c))
|
|
|
+
|
|
|
+print("扫描 git 跟踪文件数 = %d（其中读不到 %d）" % (len(tracked), missing))
|
|
|
+print("工作区版本含密钥的文件数 = %d" % len(bad))
|
|
|
+for f, c in bad:
|
|
|
+ print(" !! %s 命中 %d 次" % (f, c))
|
|
|
+
|
|
|
+head = subprocess.run(["git", "show", "HEAD:java/src/main/resources/application.yml"],
|
|
|
+ capture_output=True).stdout
|
|
|
+print("--- HEAD:application.yml ---")
|
|
|
+print("bytes = %d, 命中 = %d" % (len(head), head.count(needle)))
|
|
|
+
|
|
|
+if "--ctx" not in sys.argv:
|
|
|
+ raise SystemExit(0)
|
|
|
+
|
|
|
+# 定位版：把每个命中处的**整行**打出来（该行里的密钥替换成 <KEY>）
|
|
|
+print()
|
|
|
+print("=== 命中行定位（密钥已整段替换成 <KEY>）===")
|
|
|
+for f, _c in bad:
|
|
|
+ if f == YML:
|
|
|
+ continue # 工作区那份单独打在下面
|
|
|
+ blob = io.open(f, "rb").read()
|
|
|
+ print("--- %s (%d B)" % (f, len(blob)))
|
|
|
+ start, k = 0, 0
|
|
|
+ while True:
|
|
|
+ i = blob.find(needle, start)
|
|
|
+ if i < 0:
|
|
|
+ break
|
|
|
+ k += 1
|
|
|
+ ls = blob.rfind(b"\n", 0, i) + 1
|
|
|
+ le = blob.find(b"\n", i)
|
|
|
+ if le < 0:
|
|
|
+ le = len(blob)
|
|
|
+ line = blob[ls:le].decode("utf-8", "replace").replace(KEY, "<KEY>")
|
|
|
+ print(" #%d 第 %d 行: %s" % (k, blob.count(b"\n", 0, ls) + 1, line.strip()))
|
|
|
+ start = i + len(needle)
|
|
|
+
|
|
|
+for label, blob in (("工作区 application.yml", io.open(YML, "rb").read()),
|
|
|
+ ("HEAD:application.yml", head)):
|
|
|
+ i = blob.find(needle)
|
|
|
+ if i < 0:
|
|
|
+ continue
|
|
|
+ ls = blob.rfind(b"\n", 0, i) + 1
|
|
|
+ le = blob.find(b"\n", i)
|
|
|
+ if le < 0:
|
|
|
+ le = len(blob)
|
|
|
+ print("--- %s 第 %d 行: %s"
|
|
|
+ % (label, blob.count(b"\n", 0, ls) + 1,
|
|
|
+ blob[ls:le].decode("utf-8", "replace").replace(KEY, "<KEY>").strip()))
|
alphaH