85 lines
3.0 KiB
Python
85 lines
3.0 KiB
Python
"""可信临时包扫描器；只解析 dist-info 元数据，不 import 用户代码。"""
|
|
from __future__ import annotations
|
|
import csv
|
|
import email.parser
|
|
import json
|
|
from pathlib import Path
|
|
|
|
BASE = Path("/sandbox/base-python-packages.json")
|
|
ROOT = Path("/tmp/.local/lib")
|
|
|
|
|
|
def safe_file(root: Path, path: Path) -> bool:
|
|
try:
|
|
path.relative_to(root)
|
|
except ValueError:
|
|
return False
|
|
current = root
|
|
for part in path.relative_to(root).parts:
|
|
current = current / part
|
|
if current.is_symlink():
|
|
return False
|
|
return path.is_file()
|
|
|
|
|
|
def safe_dir(root: Path, path: Path) -> bool:
|
|
try:
|
|
rel = path.relative_to(root)
|
|
except ValueError:
|
|
return False
|
|
current = root
|
|
for part in rel.parts:
|
|
current = current / part
|
|
if current.is_symlink():
|
|
return False
|
|
return path.is_dir()
|
|
|
|
|
|
def canonical(name: str) -> str:
|
|
return name.lower().replace("_", "-").replace(".", "-")
|
|
|
|
|
|
def scan_packages(root: Path, base: dict) -> list[dict]:
|
|
out = []
|
|
if root.is_dir():
|
|
for site in sorted(root.glob("python*/site-packages")):
|
|
if not safe_dir(root, site):
|
|
continue
|
|
for dist in sorted(site.glob("*.dist-info")):
|
|
if len(out) >= 5000:
|
|
return out
|
|
meta_path = dist / "METADATA"
|
|
if not safe_file(site, meta_path) or meta_path.stat().st_size > 1024 * 1024:
|
|
continue
|
|
msg = email.parser.Parser().parsestr(meta_path.read_text(encoding="utf-8", errors="replace"), headersonly=True)
|
|
name, version = msg.get("Name", "").strip(), msg.get("Version", "").strip()
|
|
if not name:
|
|
continue
|
|
total = 0
|
|
record = dist / "RECORD"
|
|
if safe_file(site, record) and record.stat().st_size <= 32 * 1024 * 1024:
|
|
for index, row in enumerate(csv.reader(record.read_text(encoding="utf-8", errors="replace").splitlines())):
|
|
if index >= 200_000:
|
|
break
|
|
if not row:
|
|
continue
|
|
candidate = site / row[0]
|
|
if safe_file(site, candidate):
|
|
try: total += candidate.stat().st_size
|
|
except OSError: pass
|
|
key = canonical(name)
|
|
old = base.get(key)
|
|
change = "added" if old is None else ("reinstall" if old == version else "override")
|
|
out.append({"name": name, "version": version, "base_version": old,
|
|
"change": change, "direct_requested": safe_file(site, dist / "REQUESTED") or safe_file(site, dist / "direct_url.json"),
|
|
"installed_bytes": total})
|
|
return out
|
|
|
|
|
|
def main() -> None:
|
|
base = json.loads(BASE.read_text(encoding="utf-8"))
|
|
print(json.dumps(scan_packages(ROOT, base), ensure_ascii=False))
|
|
|
|
if __name__ == "__main__":
|
|
main()
|