payment-platform/logs/_lsy-repo-keyscan.py

# -*- 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()))