257 lines
8.8 KiB
Python
257 lines
8.8 KiB
Python
"""What must never reach the pool.
|
|
|
|
AGENTS.md §4.4 and `docs/phase-4-corpus.md` §6 make this a hard gate, not a
|
|
suggestion: node_modules, virtualenvs, git internals, `.trash`, `.obsidian`, the
|
|
SilverBullet auth file, credential-shaped files and binaries are refused. The
|
|
platform guide excludes are *additional*; they cannot widen this list.
|
|
|
|
Two different outcomes, because they are different things:
|
|
|
|
* **reject** — the file is not read at all. The reason is recorded and reported
|
|
so a run says *why* it skipped something.
|
|
* **image** — a media file is recorded as an image node reference (path only,
|
|
never bytes, never extracted text). Phase 4 §1.
|
|
|
|
Credential-shaped *content* is a third thing: a file with a valid path can still
|
|
contain a private key or an API token. `scan_text` rejects the whole document
|
|
rather than redacting part of it, and the same scan runs over the finished pool
|
|
so "no secret in the pool" is verified by search, not asserted.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
|
|
from . import globs
|
|
|
|
# Directory names that are never descended into. Checked per path component, so
|
|
# a file deep inside one is refused by the same rule as its parent.
|
|
EXCLUDED_DIRS = frozenset(
|
|
{
|
|
"node_modules",
|
|
".venv",
|
|
"venv",
|
|
".git",
|
|
".trash",
|
|
".obsidian",
|
|
".pi",
|
|
"__pycache__",
|
|
".mypy_cache",
|
|
".pytest_cache",
|
|
".ruff_cache",
|
|
"graphify-out",
|
|
"graphify-out-temp",
|
|
".next",
|
|
".cache",
|
|
"site-packages",
|
|
}
|
|
)
|
|
|
|
# Basenames (and simple suffix patterns) that are credential-shaped by name.
|
|
CREDENTIAL_FILES = (
|
|
".silverbullet.auth.json",
|
|
".env",
|
|
".env.*",
|
|
"*.env",
|
|
"*.env.*",
|
|
".netrc",
|
|
".git-credentials",
|
|
".npmrc",
|
|
".pypirc",
|
|
"credentials",
|
|
"credentials.*",
|
|
"secrets.*",
|
|
"secret.*",
|
|
"auth.json",
|
|
"*.pem",
|
|
"*.key",
|
|
"*.p12",
|
|
"*.pfx",
|
|
"*.jks",
|
|
"*.keystore",
|
|
"*.kdbx",
|
|
"*.age",
|
|
"id_rsa",
|
|
"id_ed25519",
|
|
"id_ecdsa",
|
|
"*.lock",
|
|
"*.lockb",
|
|
)
|
|
|
|
# Media is never text. It becomes an image node reference and nothing else.
|
|
IMAGE_EXTENSIONS = frozenset(
|
|
{"png", "jpg", "jpeg", "gif", "webp", "bmp", "tif", "tiff", "heic", "avif", "ico"}
|
|
)
|
|
VIDEO_EXTENSIONS = frozenset({"mp4", "mov", "mkv", "webm", "avi", "m4v", "mpg", "mpeg"})
|
|
AUDIO_EXTENSIONS = frozenset({"mp3", "wav", "flac", "ogg", "m4a", "aac", "opus", "wma"})
|
|
MEDIA_EXTENSIONS = IMAGE_EXTENSIONS | VIDEO_EXTENSIONS | AUDIO_EXTENSIONS
|
|
|
|
# Everything else that is bytes, not prose. Rejected with a reason; not indexed
|
|
# and not recorded as media.
|
|
BINARY_EXTENSIONS = frozenset(
|
|
{
|
|
"pdf", "zip", "tar", "gz", "bz2", "xz", "7z", "rar", "iso", "dmg",
|
|
"exe", "dll", "so", "dylib", "o", "a", "class", "jar", "war",
|
|
"pyc", "pyo", "wasm", "ttf", "otf", "woff", "woff2", "eot",
|
|
"sqlite", "sqlite3", "db", "mdb", "bin", "dat", "pkl", "npy", "npz",
|
|
"stl", "step", "obj", "blend", "psd", "ai", "sketch", "fig",
|
|
}
|
|
)
|
|
|
|
# Text-ish extensions the estate actually contains. A file with an unknown
|
|
# extension is admitted only when it decodes as UTF-8 and does not look binary.
|
|
TEXT_EXTENSIONS = frozenset(
|
|
{
|
|
"md", "markdown", "mdx", "txt", "rst", "adoc", "org",
|
|
"py", "go", "ts", "tsx", "js", "jsx", "mjs", "cjs", "rs", "rb",
|
|
"php", "kt", "kts", "java", "cs", "c", "h", "cc", "cpp", "hpp",
|
|
"ino", "lua", "pl", "r", "scala", "swift", "dart", "ex", "exs",
|
|
"sh", "bash", "zsh", "fish", "ps1", "bat",
|
|
"nix", "toml", "yml", "yaml", "ini", "conf", "cfg", "properties",
|
|
"json", "jsonc", "xml", "html", "htm", "css", "scss", "sass", "less",
|
|
"sql", "graphql", "proto", "tf", "hcl", "dockerfile", "makefile",
|
|
"gitignore", "dockerignore", "editorconfig", "service", "timer",
|
|
"csv", "tsv", "svg", "tex", "patch", "diff", "env.example",
|
|
"example", "sample", "template", "j2", "tmpl",
|
|
}
|
|
)
|
|
|
|
_CREDENTIAL_PATTERNS = (
|
|
("private-key", re.compile(r"-----BEGIN (?:[A-Z]+ )?PRIVATE KEY-----")),
|
|
("aws-access-key", re.compile(r"\bAKIA[0-9A-Z]{16}\b")),
|
|
("github-token", re.compile(r"\bgh[pousr]_[A-Za-z0-9]{20,}\b")),
|
|
("slack-token", re.compile(r"\bxox[baprs]-[A-Za-z0-9-]{10,}\b")),
|
|
("openai-key", re.compile(r"\bsk-[A-Za-z0-9]{20,}\b")),
|
|
("anthropic-key", re.compile(r"\bsk-ant-[A-Za-z0-9_\-]{20,}\b")),
|
|
("jwt", re.compile(r"\beyJ[A-Za-z0-9_\-]{10,}\.[A-Za-z0-9_\-]{10,}\.[A-Za-z0-9_\-]{10,}\b")),
|
|
("age-key", re.compile(r"\bAGE-SECRET-KEY-1[0-9A-Z]{20,}\b")),
|
|
(
|
|
"assigned-secret",
|
|
re.compile(
|
|
r"(?im)^\s*(?:export\s+)?[A-Z0-9_]*(?:SECRET|TOKEN|PASSWORD|PASSWD|API_KEY|"
|
|
r"ACCESS_KEY|PRIVATE_KEY)[A-Z0-9_]*\s*[:=]\s*[\"']?[^\s\"'#]{8,}"
|
|
),
|
|
),
|
|
)
|
|
|
|
# A `.env.example`-style file may document the *name* of a variable. It may not
|
|
# carry a value. This filter keeps `GITEA_TOKEN=` and `TOKEN=changeme` out.
|
|
_PLACEHOLDERS = {"", "changeme", "change-me", "xxx", "todo", "placeholder", "your-token-here", "redacted"}
|
|
|
|
|
|
class Rejection:
|
|
"""One file that was not admitted, and the rule that refused it."""
|
|
|
|
__slots__ = ("path", "platform", "reason", "detail")
|
|
|
|
def __init__(self, path: str, platform: str, reason: str, detail: str = "") -> None:
|
|
self.path = path
|
|
self.platform = platform
|
|
self.reason = reason
|
|
self.detail = detail
|
|
|
|
def to_dict(self) -> dict:
|
|
return {
|
|
"path": self.path,
|
|
"platform": self.platform,
|
|
"reason": self.reason,
|
|
"detail": self.detail,
|
|
}
|
|
|
|
|
|
def extension(path: str) -> str:
|
|
base = path.rsplit("/", 1)[-1].lower()
|
|
if "." not in base:
|
|
return base
|
|
return base.rsplit(".", 1)[-1]
|
|
|
|
|
|
def basename(path: str) -> str:
|
|
return path.rsplit("/", 1)[-1]
|
|
|
|
|
|
def classify(path: str, guide=None, size: int | None = None) -> tuple:
|
|
"""Return `(kind, reason)` for a path.
|
|
|
|
`kind` is one of `text`, `image`, `reject`. `reason` is empty for text and
|
|
names the rule for the other two. Guide excludes are applied here so a
|
|
caller cannot forget them.
|
|
"""
|
|
path = globs.normalise(path)
|
|
if not path:
|
|
return "reject", "empty path"
|
|
|
|
parts = path.split("/")
|
|
for part in parts[:-1]:
|
|
if part.lower() in EXCLUDED_DIRS:
|
|
return "reject", f"excluded directory {part!r}"
|
|
|
|
name = basename(path)
|
|
lowered = name.lower()
|
|
if lowered in EXCLUDED_DIRS:
|
|
return "reject", f"excluded directory {lowered!r}"
|
|
for pattern in CREDENTIAL_FILES:
|
|
if globs.matches(lowered, pattern) or globs.matches(name, pattern):
|
|
return "reject", "credential-shaped file name"
|
|
|
|
if guide is not None:
|
|
exclude = getattr(guide, "exclude", {}) or {}
|
|
if globs.under_any(path, exclude.get("paths")):
|
|
return "reject", "excluded by platform guide"
|
|
if globs.matches_any(path, exclude.get("files") or ()):
|
|
return "reject", "excluded by platform guide"
|
|
|
|
ext = extension(path)
|
|
if ext in MEDIA_EXTENSIONS:
|
|
return "image", f"media ({ext}) recorded as an image reference"
|
|
if ext in BINARY_EXTENSIONS:
|
|
return "reject", f"binary ({ext})"
|
|
|
|
if size is not None:
|
|
limit = int(((getattr(guide, "limits", {}) or {}).get("max_file_kb") or 200) * 1024)
|
|
if size > limit:
|
|
return "reject", f"oversize ({size} bytes > {limit})"
|
|
|
|
if guide is not None:
|
|
include = getattr(guide, "include", {}) or {}
|
|
include_paths = include.get("paths") or []
|
|
if include_paths and not globs.under_any(path, include_paths):
|
|
return "reject", "not under an included path"
|
|
|
|
if ext not in TEXT_EXTENSIONS and name.lower() not in TEXT_EXTENSIONS:
|
|
# Unknown extension: admitted only if the caller proves it is UTF-8 text.
|
|
return "unknown", ""
|
|
return "text", ""
|
|
|
|
|
|
def looks_binary(data: bytes) -> bool:
|
|
if b"\x00" in data[:4096]:
|
|
return True
|
|
try:
|
|
data.decode("utf-8")
|
|
except UnicodeDecodeError:
|
|
return True
|
|
return False
|
|
|
|
|
|
def scan_text(text: str) -> list:
|
|
"""Return `[(kind, line_number, excerpt)]` for credential-shaped content.
|
|
|
|
The excerpt is short and never includes the matched value: a rejection
|
|
report must not become the leak it is reporting.
|
|
"""
|
|
hits = []
|
|
for number, line in enumerate(text.splitlines(), start=1):
|
|
for kind, pattern in _CREDENTIAL_PATTERNS:
|
|
match = pattern.search(line)
|
|
if not match:
|
|
continue
|
|
value = match.group(0)
|
|
if kind == "assigned-secret":
|
|
assigned = value.split("=", 1)[-1].split(":", 1)[-1].strip().strip("\"'")
|
|
if assigned.lower() in _PLACEHOLDERS:
|
|
continue
|
|
hits.append((kind, number, line.strip()[:60].split("=", 1)[0]))
|
|
break
|
|
return hits
|