328 lines
12 KiB
Python
328 lines
12 KiB
Python
from __future__ import annotations
|
||
|
||
import argparse
|
||
import ipaddress
|
||
import json
|
||
import os
|
||
import re
|
||
import subprocess
|
||
import sys
|
||
from pathlib import Path, PurePosixPath
|
||
|
||
|
||
SCRIPT_PATH = Path(__file__).resolve()
|
||
WORKSPACE_ROOT = SCRIPT_PATH.parents[2]
|
||
PRIVATE_CREDENTIAL = PurePosixPath("测试相关资料/核桃派的用户名和密码和ip/用户名密码ip.txt")
|
||
EXAMPLE_CREDENTIAL = PurePosixPath(
|
||
"测试相关资料/核桃派的用户名和密码和ip/用户名密码ip.example.txt"
|
||
)
|
||
PRIVATE_MOBILE_REGISTRATION = PurePosixPath(
|
||
"移动端相关内容/安卓app/如何安卓测试/测试设备登记/测试设备.local.json"
|
||
)
|
||
EXAMPLE_MOBILE_REGISTRATION = PRIVATE_MOBILE_REGISTRATION.with_name("测试设备.example.json")
|
||
BINARY_EXTENSIONS = {
|
||
".apk",
|
||
".aab",
|
||
".jar",
|
||
".deb",
|
||
".dll",
|
||
".docx",
|
||
".exe",
|
||
".gif",
|
||
".gz",
|
||
".img",
|
||
".jpeg",
|
||
".jpg",
|
||
".otf",
|
||
".ota",
|
||
".pdf",
|
||
".png",
|
||
".rar",
|
||
".so",
|
||
".ttf",
|
||
".whl",
|
||
".woff",
|
||
".woff2",
|
||
".zip",
|
||
}
|
||
PRIVATE_IP_ALLOWED_PREFIXES = (
|
||
"整体开发需求/",
|
||
"测试相关资料/如何测试/",
|
||
"核桃派软件源代码/",
|
||
"发布更新相关/其他依赖/",
|
||
)
|
||
SAFE_SECRET_VALUES = {
|
||
"changeme",
|
||
"example",
|
||
"example123",
|
||
"fake",
|
||
"fake-secret",
|
||
"password",
|
||
"secret123",
|
||
"test",
|
||
"test-password",
|
||
}
|
||
# The password is a high-confidence private marker. Addresses, usernames and
|
||
# hostnames are checked structurally because common fixture values also occur in
|
||
# provisioning source and tests.
|
||
CURRENT_DEVICE_KEYS = ("密码",)
|
||
PUBLIC_IMAGE_KEYS = (
|
||
"镜像默认用户名",
|
||
"镜像默认账户密码",
|
||
"镜像默认WiFi SSID",
|
||
"镜像默认WiFi密码",
|
||
)
|
||
|
||
|
||
class HygieneError(RuntimeError):
|
||
"""Repository state cannot be audited safely."""
|
||
|
||
|
||
def _git(*arguments: str, check: bool = True) -> subprocess.CompletedProcess[bytes]:
|
||
return subprocess.run(
|
||
("git", *arguments),
|
||
cwd=WORKSPACE_ROOT,
|
||
stdout=subprocess.PIPE,
|
||
stderr=subprocess.PIPE,
|
||
check=check,
|
||
)
|
||
|
||
|
||
def _decode_paths(payload: bytes) -> list[PurePosixPath]:
|
||
return [
|
||
PurePosixPath(item.decode("utf-8", errors="surrogateescape"))
|
||
for item in payload.split(b"\0")
|
||
if item
|
||
]
|
||
|
||
|
||
def candidate_paths(staged: bool) -> list[PurePosixPath]:
|
||
if staged:
|
||
result = _git("diff", "--cached", "--name-only", "--diff-filter=ACMR", "-z")
|
||
return _decode_paths(result.stdout)
|
||
tracked = _decode_paths(_git("ls-files", "-z").stdout)
|
||
untracked = _decode_paths(_git("ls-files", "--others", "--exclude-standard", "-z").stdout)
|
||
return sorted(set((*tracked, *untracked)), key=str)
|
||
|
||
|
||
def _is_binary(path: Path) -> bool:
|
||
if path.suffix.casefold() in BINARY_EXTENSIONS:
|
||
return True
|
||
try:
|
||
return b"\0" in path.read_bytes()[:8192]
|
||
except OSError as error:
|
||
raise HygieneError(f"无法读取候选文件：{path.relative_to(WORKSPACE_ROOT)}") from error
|
||
|
||
|
||
def _host_text_patterns() -> list[re.Pattern[str]]:
|
||
windows_prefix = r"[A-Za-z]:[\\/]" + r"(?:Users|Documents and Settings)[\\/]"
|
||
mac_prefix = r"/" + r"Users/[A-Za-z0-9._-]+/"
|
||
linux_home = r"/" + r"home/[A-Za-z0-9._-]+/"
|
||
patterns = [re.compile(windows_prefix, re.IGNORECASE), re.compile(mac_prefix), re.compile(linux_home)]
|
||
for value in (Path.home().name, os.environ.get("COMPUTERNAME", "")):
|
||
if value and len(value) >= 4:
|
||
patterns.append(re.compile(re.escape(value), re.IGNORECASE))
|
||
return patterns
|
||
|
||
|
||
def _private_value_markers() -> tuple[str, ...]:
|
||
path = WORKSPACE_ROOT.joinpath(*PRIVATE_CREDENTIAL.parts)
|
||
if not path.is_file():
|
||
return _mobile_value_markers()
|
||
fields: dict[str, str] = {}
|
||
for raw_line in path.read_text(encoding="utf-8").splitlines():
|
||
line = raw_line.strip()
|
||
if not line or line.startswith("#"):
|
||
continue
|
||
separator = "：" if "：" in line else ":" if ":" in line else None
|
||
if separator is None:
|
||
continue
|
||
key, value = (part.strip() for part in line.split(separator, 1))
|
||
fields[key] = value
|
||
public_values = {fields.get(key, "") for key in PUBLIC_IMAGE_KEYS}
|
||
return _mobile_value_markers() + tuple(
|
||
value
|
||
for key in CURRENT_DEVICE_KEYS
|
||
if len(value := fields.get(key, "")) >= 4 and value not in public_values
|
||
)
|
||
|
||
|
||
def _mobile_value_markers() -> tuple[str, ...]:
|
||
path = WORKSPACE_ROOT.joinpath(*PRIVATE_MOBILE_REGISTRATION.parts)
|
||
if not path.is_file():
|
||
return ()
|
||
try:
|
||
document = json.loads(path.read_text(encoding="utf-8"))
|
||
devices = document["devices"]
|
||
if not isinstance(devices, list):
|
||
raise ValueError
|
||
values = []
|
||
for device in devices:
|
||
if not isinstance(device, dict):
|
||
raise ValueError
|
||
serial = device.get("serial", "")
|
||
if not isinstance(serial, str):
|
||
raise ValueError
|
||
if serial and "<" not in serial and len(serial) >= 4:
|
||
values.append(serial)
|
||
return tuple(values)
|
||
except (OSError, UnicodeError, ValueError, KeyError, TypeError):
|
||
raise HygieneError("移动端真实登记损坏，无法安全提取私有标识；未输出内容。") from None
|
||
|
||
|
||
def _binary_markers(private_values: tuple[str, ...] = ()) -> list[bytes]:
|
||
markers: list[bytes] = []
|
||
host_values = {
|
||
str(Path.home()),
|
||
str(WORKSPACE_ROOT),
|
||
Path.home().name,
|
||
os.environ.get("COMPUTERNAME", ""),
|
||
}
|
||
for value in (*host_values, *private_values):
|
||
if value and len(value) >= 4:
|
||
variants = {value, value.replace("\\", "/")}
|
||
for variant in variants:
|
||
markers.extend((variant.encode("utf-8"), variant.encode("utf-16le")))
|
||
return markers
|
||
|
||
|
||
def _binary_contains_forbidden_marker(path: Path, private_values: tuple[str, ...]) -> bool:
|
||
markers = _binary_markers(private_values)
|
||
overlap = max(map(len, markers)) - 1
|
||
tail = b""
|
||
with path.open("rb") as stream:
|
||
while chunk := stream.read(8 * 1024 * 1024):
|
||
sample = tail + chunk
|
||
lowered_sample = sample.lower()
|
||
if any(marker.lower() in lowered_sample for marker in markers):
|
||
return True
|
||
tail = sample[-overlap:] if overlap else b""
|
||
return False
|
||
|
||
|
||
def _text_issues(
|
||
relative: PurePosixPath,
|
||
text: str,
|
||
private_values: tuple[str, ...] = (),
|
||
) -> list[str]:
|
||
issues: list[str] = []
|
||
if any(pattern.search(text) for pattern in _host_text_patterns()):
|
||
issues.append("包含开发电脑专属路径、用户名或主机名")
|
||
if any(value in text for value in private_values):
|
||
issues.append("包含当前设备私有凭据或测试手机标识")
|
||
|
||
key_header = re.compile("-----BEGIN " + r"(?:OPENSSH |RSA |EC |DSA )?PRIVATE KEY-----")
|
||
if key_header.search(text):
|
||
issues.append("包含私钥正文")
|
||
|
||
relative_string = relative.as_posix()
|
||
if not (
|
||
relative_string.startswith("核桃派软件源代码/tests/")
|
||
or PurePosixPath(relative_string).name.startswith("test_")
|
||
or relative == EXAMPLE_CREDENTIAL
|
||
):
|
||
assignment = re.compile(
|
||
r"(?i)(?:password|passwd|token|secret|api[_-]?key|密码)\s*[:=]\s*[\"']([^\"']{4,})[\"']"
|
||
)
|
||
for match in assignment.finditer(text):
|
||
if match.group(1).casefold() not in SAFE_SECRET_VALUES:
|
||
issues.append("包含疑似硬编码秘密")
|
||
break
|
||
|
||
if not (
|
||
relative_string.startswith(PRIVATE_IP_ALLOWED_PREFIXES)
|
||
or "/tests/" in f"/{relative_string}"
|
||
):
|
||
for token in re.findall(r"(?<![0-9])(?:[0-9]{1,3}\.){3}[0-9]{1,3}(?![0-9])", text):
|
||
try:
|
||
address = ipaddress.ip_address(token)
|
||
except ValueError:
|
||
continue
|
||
if address.is_private and not address.is_loopback and not address.is_unspecified:
|
||
issues.append("归档或普通文档包含私有 IPv4 地址")
|
||
break
|
||
return issues
|
||
|
||
|
||
def audit_paths(paths: list[PurePosixPath], *, scan_binary: bool) -> list[tuple[str, str]]:
|
||
findings: list[tuple[str, str]] = []
|
||
private_values = _private_value_markers()
|
||
for relative in paths:
|
||
if relative in (PRIVATE_CREDENTIAL, PRIVATE_MOBILE_REGISTRATION):
|
||
findings.append((relative.as_posix(), "真实凭据进入 Git 候选集合"))
|
||
continue
|
||
if relative.suffix.lower() in {".jks", ".keystore"}:
|
||
findings.append((relative.as_posix(), "签名密钥文件进入 Git 候选集合"))
|
||
continue
|
||
path = WORKSPACE_ROOT.joinpath(*relative.parts)
|
||
if not path.is_file():
|
||
continue
|
||
if _is_binary(path):
|
||
if scan_binary and _binary_contains_forbidden_marker(path, private_values):
|
||
findings.append((relative.as_posix(), "二进制包含开发电脑标识、主目录路径或当前设备秘密"))
|
||
continue
|
||
try:
|
||
text = path.read_text(encoding="utf-8")
|
||
except UnicodeDecodeError:
|
||
findings.append((relative.as_posix(), "文本候选不是有效 UTF-8"))
|
||
continue
|
||
findings.extend(
|
||
(relative.as_posix(), issue) for issue in _text_issues(relative, text, private_values)
|
||
)
|
||
return findings
|
||
|
||
|
||
def _check_repository_contract(paths: list[PurePosixPath], staged: bool) -> list[tuple[str, str]]:
|
||
findings: list[tuple[str, str]] = []
|
||
ignored = _git("check-ignore", "--quiet", "--", PRIVATE_CREDENTIAL.as_posix(), check=False)
|
||
if ignored.returncode != 0:
|
||
findings.append((PRIVATE_CREDENTIAL.as_posix(), "真实凭据未被 .gitignore 精确排除"))
|
||
mobile_ignored = _git("check-ignore", "--quiet", "--", PRIVATE_MOBILE_REGISTRATION.as_posix(), check=False)
|
||
if mobile_ignored.returncode != 0:
|
||
findings.append((PRIVATE_MOBILE_REGISTRATION.as_posix(), "真实手机登记未被 .gitignore 排除"))
|
||
if staged:
|
||
mobile_example = _git("cat-file", "-e", f":{EXAMPLE_MOBILE_REGISTRATION.as_posix()}", check=False)
|
||
if mobile_example.returncode != 0:
|
||
findings.append((EXAMPLE_MOBILE_REGISTRATION.as_posix(), "手机空白登记示例不在 Git 索引"))
|
||
elif EXAMPLE_MOBILE_REGISTRATION not in paths:
|
||
findings.append((EXAMPLE_MOBILE_REGISTRATION.as_posix(), "手机空白登记示例不在 Git 候选集合"))
|
||
|
||
if staged:
|
||
example_in_index = _git("cat-file", "-e", f":{EXAMPLE_CREDENTIAL.as_posix()}", check=False)
|
||
if example_in_index.returncode != 0:
|
||
findings.append((EXAMPLE_CREDENTIAL.as_posix(), "示例凭据没有进入本次提交"))
|
||
elif EXAMPLE_CREDENTIAL not in paths:
|
||
findings.append((EXAMPLE_CREDENTIAL.as_posix(), "示例凭据不在 Git 候选集合"))
|
||
|
||
return findings
|
||
|
||
|
||
def main() -> int:
|
||
parser = argparse.ArgumentParser(description="检查 Git 候选文件中的凭据、主机路径和秘密")
|
||
parser.add_argument("--staged", action="store_true", help="只检查已暂存的新建或修改文件")
|
||
parser.add_argument(
|
||
"--skip-binary-scan",
|
||
action="store_true",
|
||
help="跳过大体积二进制字节扫描，仅供快速诊断",
|
||
)
|
||
args = parser.parse_args()
|
||
try:
|
||
paths = candidate_paths(args.staged)
|
||
findings = _check_repository_contract(paths, args.staged)
|
||
findings.extend(audit_paths(paths, scan_binary=not args.skip_binary_scan))
|
||
except (HygieneError, OSError, subprocess.CalledProcessError) as error:
|
||
print(f"仓库卫生检查无法完成：{error}", file=sys.stderr)
|
||
return 2
|
||
|
||
if findings:
|
||
print("仓库卫生检查失败；以下输出只包含文件路径和问题类型：", file=sys.stderr)
|
||
for path, issue in findings:
|
||
print(f"- {path}: {issue}", file=sys.stderr)
|
||
return 1
|
||
print(f"仓库卫生检查通过：已检查 {len(paths)} 个 Git 候选文件，未输出任何秘密值。")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|