279 lines
8.7 KiB
Python
Executable File
279 lines
8.7 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Privacy-safe scanner for tracked files and optional Git object history."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import hashlib
|
|
import ipaddress
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
from typing import Iterable
|
|
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
MAX_BLOB_BYTES = 5_000_000
|
|
SENSITIVE_DOTENV = re.compile(
|
|
rb"(?im)^(?P<key>(?:JWEBOT|WECHAT)_[A-Z0-9_]*"
|
|
rb"(?:KEY|TOKEN|PASSWORD|SECRET))=(?P<value>[^\r\n#]+)$"
|
|
)
|
|
TOKEN_PATTERNS = (
|
|
(
|
|
"private_key",
|
|
re.compile(rb"-----BEGIN (?:RSA |EC |OPENSSH |DSA )?PRIVATE KEY-----"),
|
|
),
|
|
(
|
|
"api_token",
|
|
re.compile(
|
|
rb"(?i)\b(?:sk-[A-Za-z0-9_-]{16,}|gh[pousr]_[A-Za-z0-9]{20,}|"
|
|
rb"github_pat_[A-Za-z0-9_]{20,}|xox[baprs]-[A-Za-z0-9-]{10,}|"
|
|
rb"AKIA[0-9A-Z]{16})\b"
|
|
),
|
|
),
|
|
(
|
|
"bearer_token",
|
|
re.compile(rb"(?i)\bBearer\s+([A-Za-z0-9._~+/-]{20,}=*)"),
|
|
),
|
|
(
|
|
"steam_id64",
|
|
re.compile(rb"\b7656119[0-9]{10}\b"),
|
|
),
|
|
(
|
|
"email",
|
|
re.compile(rb"(?i)\b[A-Z0-9._%+-]+@[A-Z0-9.-]+\.[A-Z]{2,}\b"),
|
|
),
|
|
(
|
|
"home_path",
|
|
re.compile(rb"/(?:Users|home)/[^/\s\"']+"),
|
|
),
|
|
(
|
|
"ipv4",
|
|
re.compile(rb"(?<![0-9])(?:[0-9]{1,3}\.){3}[0-9]{1,3}(?![0-9])"),
|
|
),
|
|
)
|
|
PLACEHOLDER = re.compile(
|
|
rb"(?i)(?:^$|example|sample|placeholder|change.?me|dummy|fake|test|"
|
|
rb"redact|your[_-]|<[^>]+>|\$\{|\{\{|none|null|unset|x{4,}|0{4,})"
|
|
)
|
|
ALLOWED_EMAIL_DOMAINS = {"users.noreply.github.com", "example.com"}
|
|
ALLOWED_HOME_USERS = {"jwebot", "node", "user"}
|
|
|
|
|
|
def run(*args: str) -> bytes:
|
|
return subprocess.check_output(args, cwd=ROOT)
|
|
|
|
|
|
def has_git_repository() -> bool:
|
|
result = subprocess.run(
|
|
["git", "rev-parse", "--is-inside-work-tree"],
|
|
cwd=ROOT,
|
|
text=True,
|
|
capture_output=True,
|
|
check=False,
|
|
)
|
|
return result.returncode == 0 and result.stdout.strip() == "true"
|
|
|
|
|
|
def digest(value: bytes) -> str:
|
|
return hashlib.sha256(value).hexdigest()[:12]
|
|
|
|
|
|
def allowed_ip(raw: bytes) -> bool:
|
|
try:
|
|
value = ipaddress.ip_address(raw.decode("ascii"))
|
|
except ValueError:
|
|
return True
|
|
if value.is_loopback:
|
|
return True
|
|
documentation = (
|
|
ipaddress.ip_network("192.0.2.0/24"),
|
|
ipaddress.ip_network("198.51.100.0/24"),
|
|
ipaddress.ip_network("203.0.113.0/24"),
|
|
)
|
|
return any(value in network for network in documentation)
|
|
|
|
|
|
def findings(data: bytes) -> list[tuple[str, int, str]]:
|
|
result: list[tuple[str, int, str]] = []
|
|
lines = data.splitlines()
|
|
for line_number, line in enumerate(lines, start=1):
|
|
for category, pattern in TOKEN_PATTERNS:
|
|
for match in pattern.finditer(line):
|
|
value = match.group(1) if match.lastindex else match.group(0)
|
|
if category == "email":
|
|
decoded = value.decode("utf-8", "replace")
|
|
if decoded.endswith((".service", ".socket", ".timer")):
|
|
continue
|
|
domain = decoded.rpartition("@")[2]
|
|
if domain.casefold() in ALLOWED_EMAIL_DOMAINS:
|
|
continue
|
|
elif category == "home_path":
|
|
user = value.decode("utf-8", "replace").rsplit("/", 1)[-1]
|
|
if user in ALLOWED_HOME_USERS:
|
|
continue
|
|
elif category == "ipv4" and allowed_ip(value):
|
|
continue
|
|
result.append((category, line_number, digest(value)))
|
|
for match in SENSITIVE_DOTENV.finditer(data):
|
|
value = match.group("value").strip().strip(b"'\"")
|
|
if value and not PLACEHOLDER.search(value):
|
|
line_number = data.count(b"\n", 0, match.start()) + 1
|
|
result.append(("populated_secret", line_number, digest(value)))
|
|
return result
|
|
|
|
|
|
def tracked_paths() -> list[str]:
|
|
if not has_git_repository():
|
|
excluded = {".git", ".venv", "__pycache__", ".pytest_cache"}
|
|
return sorted(
|
|
str(path.relative_to(ROOT))
|
|
for path in ROOT.rglob("*")
|
|
if path.is_file() and not any(part in excluded for part in path.parts)
|
|
)
|
|
return [
|
|
value.decode("utf-8", "surrogateescape")
|
|
for value in run(
|
|
"git",
|
|
"ls-files",
|
|
"--cached",
|
|
"--others",
|
|
"--exclude-standard",
|
|
"-z",
|
|
).split(b"\0")
|
|
if value
|
|
]
|
|
|
|
|
|
def suspicious_path(path: str) -> bool:
|
|
value = PurePath(path)
|
|
parts = set(value.parts)
|
|
if ".dbg" in parts or "dbg" in parts:
|
|
return True
|
|
if value.name.startswith("homeserver-apikey-"):
|
|
return True
|
|
if value.suffix in {".pem", ".p12", ".pfx", ".jks", ".keystore"}:
|
|
return True
|
|
return False
|
|
|
|
|
|
class PurePath:
|
|
def __init__(self, value: str) -> None:
|
|
self.parts = tuple(part for part in value.replace("\\", "/").split("/") if part)
|
|
self.name = self.parts[-1] if self.parts else ""
|
|
self.suffix = Path(self.name).suffix.casefold()
|
|
|
|
|
|
def scan_tracked() -> list[tuple[str, str, int, str]]:
|
|
result: list[tuple[str, str, int, str]] = []
|
|
for path in tracked_paths():
|
|
if suspicious_path(path):
|
|
result.append(("suspicious_path", path, 0, digest(path.encode())))
|
|
candidate = ROOT / path
|
|
try:
|
|
if candidate.stat().st_size > MAX_BLOB_BYTES:
|
|
continue
|
|
data = candidate.read_bytes()
|
|
except OSError:
|
|
continue
|
|
if b"\0" in data[:4096]:
|
|
continue
|
|
for category, line, value_digest in findings(data):
|
|
result.append((category, path, line, value_digest))
|
|
return result
|
|
|
|
|
|
def reachable_blob_paths() -> dict[str, set[str]]:
|
|
values: dict[str, set[str]] = defaultdict(set)
|
|
for raw in run("git", "rev-list", "--objects", "--all").splitlines():
|
|
sha, separator, path = raw.partition(b" ")
|
|
if separator:
|
|
values[sha.decode()].add(path.decode("utf-8", "replace"))
|
|
return values
|
|
|
|
|
|
def scan_history() -> list[tuple[str, str, int, str]]:
|
|
paths = reachable_blob_paths()
|
|
result: list[tuple[str, str, int, str]] = []
|
|
objects = run(
|
|
"git",
|
|
"cat-file",
|
|
"--batch-all-objects",
|
|
"--batch-check=%(objectname) %(objecttype) %(objectsize)",
|
|
).decode().splitlines()
|
|
for row in objects:
|
|
sha, object_type, size = row.split()
|
|
if object_type != "blob" or int(size) > MAX_BLOB_BYTES:
|
|
continue
|
|
data = run("git", "cat-file", "blob", sha)
|
|
if b"\0" in data[:4096]:
|
|
continue
|
|
for category, line, value_digest in findings(data):
|
|
for path in sorted(paths.get(sha, {"<unreachable-or-unmapped>"})):
|
|
result.append(
|
|
(
|
|
f"history_{category}",
|
|
f"{path}@{sha[:12]}",
|
|
line,
|
|
value_digest,
|
|
)
|
|
)
|
|
identities = run(
|
|
"git",
|
|
"log",
|
|
"--all",
|
|
"--format=%an%x00%ae%x00%cn%x00%ce",
|
|
).splitlines()
|
|
for row in identities:
|
|
fields = row.decode("utf-8", "replace").split("\0")
|
|
for email in (fields[1], fields[3]):
|
|
domain = email.rpartition("@")[2].casefold()
|
|
if domain not in ALLOWED_EMAIL_DOMAINS:
|
|
result.append(
|
|
(
|
|
"history_author_email",
|
|
"<commit-metadata>",
|
|
0,
|
|
digest(email.encode()),
|
|
)
|
|
)
|
|
return result
|
|
|
|
|
|
def unique(
|
|
values: Iterable[tuple[str, str, int, str]],
|
|
) -> list[tuple[str, str, int, str]]:
|
|
return sorted(set(values))
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(
|
|
description="Scan JWeBot without printing matched secret or PII values"
|
|
)
|
|
parser.add_argument("--history", action="store_true")
|
|
args = parser.parse_args(argv)
|
|
result = scan_tracked()
|
|
if args.history:
|
|
if not has_git_repository():
|
|
print(
|
|
"check-secrets: --history requires a Git repository",
|
|
file=sys.stderr,
|
|
)
|
|
return 2
|
|
result.extend(scan_history())
|
|
result = unique(result)
|
|
for category, path, line, value_digest in result:
|
|
location = f"{path}:{line}" if line else path
|
|
print(f"{category} {location} sha256={value_digest}")
|
|
if result:
|
|
print(f"privacy_scan=failed findings={len(result)}", file=sys.stderr)
|
|
return 1
|
|
print("privacy_scan=ok findings=0")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|