#!/usr/bin/env python3 """Monument Repo Check: a local, read-only, offline repository assessment. What it does Reads one or more git repositories on your machine and writes, for each one: monument-share.json counts and fixed labels only: the one file you may choose to send monument-private-report.json the detailed report (file paths, locations of possible secrets): keep it monument-private-report.md the same detail as a readable report: keep it monument-dependencies.csv declared dependencies, a starting point for a bill of materials: keep it What it never does * makes a network connection or calls any AI service * runs code from your repository (no tests, builds, installers, hooks, diff drivers or filters) * changes your repository or writes anything inside it * prints or stores secret values (the private report shows only where a possible secret is) * puts source code, file paths, emails, people's names, commit hashes, commit messages or remote URLs in monument-share.json; a built-in leak guard refuses to write the share file if any of those appear Requirements: Python 3.8+ and git. Standard library only. Read every line before you run it. Save this file outside every repository (for example in your Downloads folder) and run it with `-I`. Usage python3 -I monument_repo_check.py /path/to/repo --out /path/outside/the/repo python3 -I monument_repo_check.py repoA repoB repoC --out ~/monument-reports python3 -I monument_repo_check.py --discover ~/code --out ~/monument-reports python3 -I monument_repo_check.py --self-test python3 -I monument_repo_check.py --help (on Windows: py -3 -I monument_repo_check.py ...) Automated scanning is an aid, not proof: a clean result never shows that a repository is free of secrets or personal data, and nothing here establishes ownership or licensing rights. """ import sys # Never import a module from this file's folder: a repository that holds `json.py`, `csv.py` or `secrets.py` must not # be able to run code if the scanner was saved inside it. `python3 -I` already leaves that folder out; this covers a # plain `python3 monument_repo_check.py`. (Nothing above this line imports a module that is not built in.) if __name__ == "__main__" and not sys.flags.isolated and not getattr(sys.flags, "safe_path", False) and sys.path: del sys.path[0] sys.path[:] = [p for p in sys.path if p not in ("", ".")] import argparse import csv import datetime as _dt import hashlib import hmac import io import json import math import os import platform import re import secrets as _secrets import shutil import stat import subprocess import sys import tempfile import time from collections import Counter, defaultdict SCANNER_NAME = "monument-repo-check" SCANNER_VERSION = "1.0.0" SHARE_SCHEMA = "monument.repo-check.share" SHARE_SCHEMA_VERSION = "1.0" BUNDLE_SCHEMA = "monument.repo-check.bundle" PRIVATE_SCHEMA = "monument.repo-check.private" INTEGRITY_ALGORITHM = "sha256-canonical-json-v1" REF_METHOD_KEYED = "hmac-sha256-local-key-v1" REF_METHOD_EPHEMERAL = "hmac-sha256-ephemeral-v1" MAX_ANALYZE_BYTES = 1024 * 1024 # files above this are counted but not read QUICK_ANALYZE_BYTES = 256 * 1024 MAX_HISTORY_COMMITS = 50000 # per-commit analysis cap (counts stay exact) QUICK_HISTORY_COMMITS = 3000 MAX_HISTORY_PATCH_BYTES = 256 * 1024 * 1024 BATCH_BYTES = 24 * 1024 * 1024 LABEL_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9 ._-]{0,47}$") AI_ERA_START = int(_dt.datetime(2023, 1, 1, tzinfo=_dt.timezone.utc).timestamp()) GIT_EPOCH = int(_dt.datetime(2005, 4, 1, tzinfo=_dt.timezone.utc).timestamp()) # --------------------------------------------------------------------------------------------------------------- # Fixed vocabularies. Everything in monument-share.json is a number, a boolean, or a value from these lists. # --------------------------------------------------------------------------------------------------------------- EXTENSIONS = { ".py": "Python", ".pyi": "Python", ".pyw": "Python", ".js": "JavaScript", ".mjs": "JavaScript", ".cjs": "JavaScript", ".jsx": "JavaScript", ".ts": "TypeScript", ".tsx": "TypeScript", ".mts": "TypeScript", ".cts": "TypeScript", ".go": "Go", ".rs": "Rust", ".java": "Java", ".kt": "Kotlin", ".kts": "Kotlin", ".scala": "Scala", ".sc": "Scala", ".swift": "Swift", ".m": "Objective-C", ".mm": "Objective-C", ".c": "C", ".h": "C", ".cc": "C++", ".cpp": "C++", ".cxx": "C++", ".hpp": "C++", ".hh": "C++", ".hxx": "C++", ".cs": "C#", ".fs": "F#", ".fsi": "F#", ".fsx": "F#", ".vb": "Visual Basic", ".rb": "Ruby", ".rake": "Ruby", ".php": "PHP", ".pl": "Perl", ".pm": "Perl", ".lua": "Lua", ".r": "R", ".jl": "Julia", ".dart": "Dart", ".ex": "Elixir", ".exs": "Elixir", ".erl": "Erlang", ".hrl": "Erlang", ".hs": "Haskell", ".ml": "OCaml", ".mli": "OCaml", ".clj": "Clojure", ".cljs": "Clojure", ".cljc": "Clojure", ".groovy": "Groovy", ".gradle": "Groovy", ".sh": "Shell", ".bash": "Shell", ".zsh": "Shell", ".ps1": "PowerShell", ".psm1": "PowerShell", ".sql": "SQL", ".html": "HTML", ".htm": "HTML", ".css": "CSS", ".scss": "CSS", ".sass": "CSS", ".less": "CSS", ".vue": "Vue", ".svelte": "Svelte", ".sol": "Solidity", ".zig": "Zig", ".nim": "Nim", ".f": "Fortran", ".for": "Fortran", ".f77": "Fortran", ".f90": "Fortran", ".f95": "Fortran", ".f03": "Fortran", ".cob": "COBOL", ".cbl": "COBOL", ".cpy": "COBOL", ".asm": "Assembly", ".s": "Assembly", ".adb": "Ada", ".ads": "Ada", ".pas": "Pascal", ".dpr": "Pascal", ".rpgle": "RPG", ".sqlrpgle": "RPG", ".abap": "ABAP", ".tf": "HCL", ".hcl": "HCL", ".proto": "Protocol Buffers", ".graphql": "GraphQL", ".gql": "GraphQL", ".cmake": "CMake", ".nix": "Nix", ".v": "Verilog", ".sv": "Verilog", ".vhd": "VHDL", ".vhdl": "VHDL", ".cr": "Crystal", ".elm": "Elm", ".bzl": "Starlark", ".ino": "C++", } FILENAMES = { "makefile": "Makefile", "gnumakefile": "Makefile", "dockerfile": "Dockerfile", "cmakelists.txt": "CMake", "jenkinsfile": "Groovy", "rakefile": "Ruby", "build.bazel": "Starlark", "workspace": "Starlark", "module.bazel": "Starlark", } NON_CODE_EXTENSIONS = { ".md": "Markdown", ".mdx": "Markdown", ".rst": "Markdown", ".txt": "Text", ".json": "Data", ".yml": "Data", ".yaml": "Data", ".toml": "Data", ".xml": "Data", ".csv": "Data", ".tsv": "Data", ".ini": "Data", ".cfg": "Data", ".lock": "Data", } LANGUAGES = sorted(set(EXTENSIONS.values()) | set(FILENAMES.values())) RARE_STACKS = {"COBOL", "Fortran", "RPG", "ABAP", "Ada", "Pascal", "Assembly", "Visual Basic"} LINE_COMMENT = { "#": ["Python", "Ruby", "Shell", "Perl", "R", "Julia", "Elixir", "Nim", "Makefile", "Dockerfile", "CMake", "PowerShell", "HCL", "Crystal", "GraphQL", "Starlark", "Nix"], "//": ["C", "C++", "C#", "Java", "JavaScript", "TypeScript", "Go", "Rust", "Kotlin", "Scala", "Swift", "Objective-C", "Dart", "PHP", "Groovy", "Solidity", "Zig", "F#", "Protocol Buffers", "Vue", "Svelte", "HCL", "Verilog", "Pascal"], "--": ["SQL", "Lua", "Haskell", "Ada", "Elm", "VHDL"], "%": ["Erlang"], ";": ["Clojure", "Assembly"], "!": ["Fortran"], "'": ["Visual Basic"], "*>": ["COBOL"], } BLOCK_COMMENT = { ("/*", "*/"): ["C", "C++", "C#", "Java", "JavaScript", "TypeScript", "Go", "Rust", "Kotlin", "Scala", "Swift", "Objective-C", "Dart", "PHP", "Groovy", "Solidity", "CSS", "SQL", "Protocol Buffers", "Vue", "Svelte", "HCL", "Verilog"], ('"""', '"""'): ["Python"], ("'''", "'''"): ["Python"], (""): ["HTML", "Vue", "Svelte"], ("{-", "-}"): ["Haskell", "Elm"], ("(*", "*)"): ["OCaml", "F#", "Pascal"], ("--[[", "]]"): ["Lua"], ("<#", "#>"): ["PowerShell"], ("=begin", "=end"): ["Ruby"], } CI_SYSTEMS = ["github_actions", "gitlab_ci", "circleci", "jenkins", "azure_pipelines", "travis_ci", "bitbucket_pipelines", "buildkite", "drone", "appveyor", "google_cloud_build", "aws_codebuild", "teamcity", "woodpecker", "semaphore"] TEST_FRAMEWORKS = ["pytest", "unittest", "nose2", "hypothesis", "jest", "vitest", "mocha", "jasmine", "ava", "playwright", "cypress", "karma", "go_test", "cargo_test", "junit", "testng", "kotest", "spock", "scalatest", "xunit", "nunit", "mstest", "rspec", "minitest", "phpunit", "pest", "xctest", "exunit", "dart_test", "flutter_test", "googletest", "catch2", "ctest", "testthat"] FRAMEWORKS = ["react", "nextjs", "vue", "nuxt", "angular", "svelte", "express", "nestjs", "fastify", "django", "flask", "fastapi", "rails", "spring", "laravel", "symfony", "aspnet", "gin", "echo", "fiber", "actix", "axum", "tokio", "pytorch", "tensorflow", "jax", "scikit_learn", "transformers", "pandas", "spark", "airflow", "dbt", "react_native", "expo", "flutter", "electron", "graphql", "prisma", "sqlalchemy", "hibernate", "langchain"] DOMAIN_TAGS = ["web_frontend", "web_backend", "mobile", "desktop", "machine_learning", "llm_applications", "data_engineering", "scientific_computing", "computer_vision", "cloud_infrastructure", "devops", "embedded", "blockchain", "fintech", "healthcare", "game_development", "security", "databases", "developer_tools"] QUALITY_TOOLS = ["linter", "formatter", "type_checker", "pre_commit", "coverage_config", "codeowners", "contributing_guide", "security_policy", "changelog", "docs_directory", "dependency_updates"] BUILD_TOOLS = ["docker", "docker_compose", "devcontainer", "nix", "makefile", "justfile", "taskfile", "bazel", "cmake", "gradle", "maven", "npm_scripts", "tox", "nox", "poetry", "uv", "pipenv", "conda", "cargo", "go_modules", "dotnet", "mix", "bundler", "composer", "sbt"] RUNTIME_PINS = ["node", "python", "ruby", "rust", "go", "dotnet", "java", "multi_tool"] WORKSPACE_TOOLS = ["pnpm_workspace", "npm_workspaces", "yarn_workspaces", "lerna", "nx", "turbo", "rush", "go_work", "cargo_workspace", "gradle_multiproject", "bazel"] ECOSYSTEMS = ["npm", "pypi", "go", "cargo", "rubygems", "maven", "gradle", "composer", "nuget", "pub", "hex"] REMOTE_HOSTING = ["github", "gitlab", "bitbucket", "azure_devops", "other_host", "local_path"] LICENCES = ["MIT", "Apache-2.0", "BSD-2-Clause", "BSD-3-Clause", "ISC", "MPL-2.0", "LGPL", "GPL-2.0", "GPL-3.0", "GPL", "AGPL-3.0", "EPL-2.0", "Unlicense", "CC0-1.0", "CC-BY", "BUSL-1.1", "SSPL-1.0", "Zlib", "proprietary", "unrecognized", "none"] LICENCE_FAMILIES = ["permissive", "weak_copyleft", "strong_copyleft", "network_copyleft", "public_domain", "source_available", "proprietary", "unknown", "none"] SECRET_KINDS = ["aws_access_key", "aws_secret_key", "github_token", "gitlab_token", "slack_token", "slack_webhook", "stripe_live_key", "google_api_key", "private_key_block", "jwt", "openai_key", "anthropic_key", "sendgrid_key", "twilio_key", "npm_token", "connection_string_password", "generic_secret_assignment"] LIMITATIONS = ["no_git_metadata", "no_commits", "shallow_history", "partial_clone", "history_sampled", "quick_mode", "history_secret_scan_skipped", "history_secret_scan_truncated", "large_files_not_analyzed", "binary_files_not_analyzed", "lfs_pointers_present", "submodules_not_scanned", "worktree_dirty_not_scanned", "scanned_ref_not_checked_out", "working_tree_without_git", "dependency_licences_not_resolved", "tests_not_executed", "secret_scan_is_heuristic", "personal_data_scan_is_heuristic"] DECLARATION_ANSWERS = ["yes", "no", "not_stated"] # --------------------------------------------------------------------------------------------------------------- # Small helpers # --------------------------------------------------------------------------------------------------------------- class ScanError(Exception): pass class LeakError(Exception): pass def now_utc(): return _dt.datetime.now(_dt.timezone.utc) def bp(part, whole): """Basis points (0..10000) as an integer: the share file carries no floats.""" if not whole or whole <= 0: return 0 return max(0, min(10000, int(round(10000.0 * part / whole)))) def months_between(t0, t1): if t0 is None or t1 is None or t1 < t0: return 0 return int((t1 - t0) // (86400 * 30.4375)) _EPOCH = _dt.datetime(1970, 1, 1, tzinfo=_dt.timezone.utc) MAX_TIMESTAMP = 253402300799 # 9999-12-31T23:59:59Z def clamp_ts(ts): """Commit timestamps come from repository content and can hold any number: keep them within 1970..9999.""" return max(0, min(int(ts), MAX_TIMESTAMP)) def utc_dt(ts): """UTC datetime for a timestamp, without the platform limits of fromtimestamp (Windows stops at year 3000).""" return _EPOCH + _dt.timedelta(seconds=clamp_ts(ts)) def calendar_months(t0, t1): """Calendar months from the first to the last timestamp (Jan 31 → Feb 1 is one month), matching the active span.""" if t0 is None or t1 is None or t1 < t0: return 0 a, b = utc_dt(t0), utc_dt(t1) return (b.year - a.year) * 12 + (b.month - a.month) def shannon(s): if not s: return 0.0 counts = Counter(s) n = float(len(s)) return -sum((c / n) * math.log2(c / n) for c in counts.values()) def canonical_json(obj): return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=True) def script_sha256(): try: with open(os.path.abspath(__file__), "rb") as fh: return hashlib.sha256(fh.read()).hexdigest() except OSError: return "0" * 64 O_BINARY = getattr(os, "O_BINARY", 0) # Windows: no newline translation, no stop at byte 0x1A def is_within(child, parent): """True when `child` is `parent` or inside it, also on case-insensitive file systems (macOS, Windows), where the same folder can be spelled differently: the folder identity (device, inode) of `parent` is looked for among the nearest existing folder of `child` and every folder above it.""" c = os.path.normcase(os.path.realpath(child)) p = os.path.normcase(os.path.realpath(parent)) if c == p or c.startswith(p.rstrip(os.sep) + os.sep): return True try: pst = os.stat(parent) except OSError: return False if not pst.st_ino: return False cur = os.path.realpath(child) while True: try: st = os.stat(cur) if (st.st_dev, st.st_ino) == (pst.st_dev, pst.st_ino): return True except OSError: pass up = os.path.dirname(cur) if up == cur: return False cur = up def write_file(path, data, private): """Write atomically; private files are created readable by the owner only.""" tmp = path + ".tmp-" + _secrets.token_hex(4) mode = 0o600 if private else 0o644 fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL | O_BINARY, mode) with os.fdopen(fd, "wb") as fh: fh.write(data.encode("utf-8") if isinstance(data, str) else data) os.replace(tmp, path) # --------------------------------------------------------------------------------------------------------------- # Git, hardened: read-only commands, nothing from the repository is executed, no network # --------------------------------------------------------------------------------------------------------------- GIT_HARDENING = [ "-c", "core.fsmonitor=false", # a repository-configured fsmonitor is a program: never run it "-c", "core.untrackedCache=false", "-c", "protocol.allow=never", # no transport at all, so no lazy fetch and no network ... ] + [x for proto in ("file", "git", "ssh", "http", "https", "ftp", "ftps", "ext", "fd") for x in ("-c", "protocol.%s.allow=never" % proto)] + [ # ... also where the repository's config allows one "-c", "core.sshCommand=false", "-c", "credential.helper=", "-c", "log.showSignature=false", # signature checks would start gpg "-c", "core.pager=cat", "-c", "color.ui=never", "-c", "core.quotePath=false", "-c", "diff.renames=false", "-c", "log.mailmap=true", ] def git_env(): env = {k: v for k, v in os.environ.items() if not k.startswith("GIT_")} env.update({ "GIT_TERMINAL_PROMPT": "0", "GIT_OPTIONAL_LOCKS": "0", "GIT_NO_LAZY_FETCH": "1", "GIT_PAGER": "cat", "PAGER": "cat", "LC_ALL": "C", "LANG": "C", "GIT_ADVICE": "0", "GIT_NO_REPLACE_OBJECTS": "1", }) return env _GIT = {"resolved": False, "bin": None, "version": (0, 0)} def resolve_git(avoid=()): """Find git by an absolute path: PATH entries that are relative ('.', '') or inside a folder being checked are skipped, so a `git` or `git.exe` placed in a repository is never run (Windows would otherwise look in the current folder first).""" names = ["git.exe"] if os.name == "nt" else ["git"] found = None for d in os.environ.get("PATH", "").split(os.pathsep): d = d.strip().strip('"') if not d or not os.path.isabs(d) or any(is_within(d, a) for a in avoid): continue for n in names: cand = os.path.join(d, n) if os.path.isfile(cand) and os.access(cand, os.X_OK): found = cand break if found: break version = (0, 0) if found: try: p = subprocess.run([found, "--version"], stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=git_env(), timeout=60) m = re.search(r"(\d+)\.(\d+)", p.stdout.decode("ascii", "replace")) if p.returncode == 0 else None if m: version = (int(m.group(1)), int(m.group(2))) else: found = None except (OSError, subprocess.SubprocessError): found = None _GIT.update({"resolved": True, "bin": found, "version": version}) return found def git_bin(): if not _GIT["resolved"]: resolve_git() return _GIT["bin"] def git_can_refuse_lazy_fetch(): return _GIT["version"] >= (2, 44) class Git: def __init__(self, repo): self.repo = repo self.env = git_env() self.calls = 0 def argv(self, args): exe = git_bin() if not exe: raise ScanError("git was not found on PATH; install git and try again") # git 2.44+ honours --no-lazy-fetch (and GIT_NO_LAZY_FETCH): a partial clone never downloads missing objects. lazy = ["--no-lazy-fetch"] if git_can_refuse_lazy_fetch() else [] return [exe] + lazy + ["-C", self.repo] + GIT_HARDENING + list(args) def run(self, args, input_bytes=None, check=True, timeout=1800): self.calls += 1 try: p = subprocess.run(self.argv(args), input=input_bytes, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=self.env, timeout=timeout) except FileNotFoundError: raise ScanError("git was not found on PATH; install git and try again") except subprocess.TimeoutExpired: raise ScanError("a git command timed out; try --quick for very large repositories") if check and p.returncode != 0: msg = p.stderr.decode("utf-8", "replace").strip().splitlines() raise ScanError("git %s failed: %s" % (args[0], msg[-1] if msg else "exit %d" % p.returncode)) return p def out(self, args, check=True): p = self.run(args, check=check) if p.returncode != 0: return None return p.stdout.decode("utf-8", "replace") def stream(self, args): self.calls += 1 return subprocess.Popen(self.argv(args), stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, env=self.env) def git_available(): return git_bin() is not None # --------------------------------------------------------------------------------------------------------------- # Path classification # --------------------------------------------------------------------------------------------------------------- VENDOR_DIRS = {"vendor", "vendors", "third_party", "third-party", "thirdparty", "3rdparty", "external", "externals", "extern", "node_modules", "bower_components", "pods", "carthage", "site-packages", ".yarn", "jspm_packages", "web_modules", "packages_vendor"} BUILD_OUTPUT_DIRS = {"dist", "build", "target", ".next", "coverage", "__pycache__", ".nuxt", ".output", "storybook-static", ".svelte-kit", ".gradle", ".tox", ".venv", "venv"} GENERATED_DIRS = {"generated", "__generated__", "autogen", ".generated", "gen-src"} GENERATED_NAME_RES = [re.compile(p) for p in [ r"_pb2(_grpc)?\.pyi?$", r"\.pb\.(go|cc|h|swift)$", r"\.pb\.gw\.go$", r"\.g\.dart$", r"\.freezed\.dart$", r"\.designer\.cs$", r"\.Designer\.cs$", r"\.g\.cs$", r"[._-]generated\.[A-Za-z]+$", r"\.min\.(js|css)$", r"\.bundle\.js$", r"_string\.go$", r"\.gen\.(go|ts|js)$", r"_gen\.go$", r"zz_generated\.[a-z.]+$", ]] GENERATED_MARKERS = ["code generated", "do not edit", "@generated", "auto-generated", "autogenerated", " 6 and raw[6] in "*/": comment += 1 continue if lang == "Fortran" and raw[:1] in ("C", "c", "*"): comment += 1 continue if prefixes and line.startswith(prefixes): comment += 1 continue opened = False for start, end in blocks: if line.startswith(start): comment += 1 if end not in line[len(start):]: end_of_block = end opened = True break if not opened: code += 1 return code, comment def normalized_hash(text): lines = [ln.strip() for ln in text.splitlines()] return hashlib.sha1("\n".join(ln for ln in lines if ln).encode("utf-8", "replace")).hexdigest() # --------------------------------------------------------------------------------------------------------------- # Secret and personal-data rules. Values are never stored, printed or written: only rule ids and locations. # --------------------------------------------------------------------------------------------------------------- SECRET_RULES = [ ("aws_access_key", "high", re.compile(r"\b(?:AKIA|ASIA)[0-9A-Z]{16}\b"), ("AKIA", "ASIA")), ("aws_secret_key", "high", re.compile(r"(?i)aws_?secret_?access_?key\s*[=:]\s*[\"']?([A-Za-z0-9/+=]{40})"), ("secret_access", "secretaccess", "SECRET_ACCESS")), ("github_token", "high", re.compile(r"\b(?:gh[pousr]_[A-Za-z0-9]{36,255}|github_pat_[A-Za-z0-9_]{22,255})\b"), ("ghp_", "gho_", "ghu_", "ghs_", "ghr_", "github_pat_")), ("gitlab_token", "high", re.compile(r"\bglpat-[A-Za-z0-9_\-]{20,}\b"), ("glpat-",)), ("slack_token", "high", re.compile(r"\bxox[abprs]-[A-Za-z0-9-]{10,}\b"), ("xox",)), ("slack_webhook", "high", re.compile(r"https://hooks\.slack\.com/services/[A-Za-z0-9/_-]{20,}"), ("hooks.slack.com",)), ("stripe_live_key", "high", re.compile(r"\b[rs]k_live_[A-Za-z0-9]{20,}\b"), ("k_live_",)), ("google_api_key", "high", re.compile(r"\bAIza[0-9A-Za-z_\-]{35}\b"), ("AIza",)), ("private_key_block", "high", re.compile(r"-----BEGIN (?:RSA |EC |DSA |OPENSSH |PGP |ENCRYPTED )?PRIVATE KEY(?: BLOCK)?-----"), ("PRIVATE KEY",)), ("anthropic_key", "high", re.compile(r"\bsk-ant-[A-Za-z0-9_\-]{32,}\b"), ("sk-ant-",)), ("openai_key", "high", re.compile(r"\bsk-(?:proj-|svcacct-)?[A-Za-z0-9_\-]{40,}\b"), ("sk-",)), ("sendgrid_key", "high", re.compile(r"\bSG\.[A-Za-z0-9_\-]{22}\.[A-Za-z0-9_\-]{43}\b"), ("SG.",)), ("twilio_key", "medium", re.compile(r"\bSK[0-9a-f]{32}\b"), ("SK",)), ("npm_token", "high", re.compile(r"\bnpm_[A-Za-z0-9]{36}\b"), ("npm_",)), ("connection_string_password", "high", re.compile(r"\b(?:postgres(?:ql)?|mysql|mariadb|mongodb(?:\+srv)?|redis|rediss|amqps?|mssql)://[^\s:/@\"']+:([^\s@/\"']{4,})@"), ("://",)), ("jwt", "medium", re.compile(r"\beyJ[A-Za-z0-9_\-]{10,}\.eyJ[A-Za-z0-9_\-]{10,}\.[A-Za-z0-9_\-]{10,}"), ("eyJ",)), ("generic_secret_assignment", "medium", re.compile(r"(?i)\b(?:password|passwd|pwd|secret|api[_-]?key|access[_-]?token|auth[_-]?token|client[_-]?secret|private[_-]?key)\b[\"']?\s*[:=]\s*[\"']([^\"'\s]{8,128})[\"']"), ("password", "passwd", "pwd", "secret", "key", "token", "PASSWORD", "SECRET", "KEY", "TOKEN", "Password", "Secret", "Key", "Token")), ] # AWS's own documentation keys (assembled so this file never holds a key-shaped literal). DOCUMENTED_EXAMPLES = {"AKIA" + "IOSFODNN7EXAMPLE", "wJalrXUtnFEMI/K7MDENG/" + "bPxRfiCYEXAMPLEKEY"} PLACEHOLDER_HINTS = ("example", "changeme", "change_me", "xxxx", "your", "dummy", "sample", "placeholder", "fake", "redacted", "<", "${", "{{", "process.env", "os.environ", "getenv", "env(", "secret_here", "password123", "*****", "todo", "replace", "insert", "test", "none", "null") EMAIL_RE = re.compile(r"(? 9: d -= 9 total += d alt = not alt return total % 10 == 0 def secret_value_key(rule, value): return hashlib.sha256((rule + "\0" + value).encode("utf-8", "replace")).hexdigest() def find_secrets(text): """Yield (rule, severity, line_no, value_key, documented_example). The value itself never leaves this function.""" for rule, severity, rx, hints in SECRET_RULES: if not any(h in text for h in hints): continue for m in rx.finditer(text): value = m.group(1) if m.groups() else m.group(0) if rule == "generic_secret_assignment": low = value.lower() if any(h in low for h in PLACEHOLDER_HINTS) or shannon(value) < 3.0 or value.isalpha(): continue if rule == "openai_key" and value.startswith("sk-ant-"): continue documented = value in DOCUMENTED_EXAMPLES or "EXAMPLE" in value line_no = text.count("\n", 0, m.start()) + 1 yield rule, severity, line_no, secret_value_key(rule, value), documented def find_personal_data(text, is_test): emails = set() if "@" in text: for m in EMAIL_RE.finditer(text): e = m.group(0).lower() dom = e.rsplit("@", 1)[1] if e.startswith("git@") or any(dom == d.lstrip(".") or dom.endswith(d if d.startswith(".") else "." + d) for d in EMAIL_IGNORE_DOMAINS): continue if re.search(r"\.(png|jpe?g|gif|svg|webp|js|css|ts)$", e): continue # retina image names such as icon@2x.png emails.add(hashlib.sha256(e.encode()).hexdigest()) ids = 0 if "-" in text: ids = sum(1 for _ in NATIONAL_ID_RE.finditer(text)) cards = 0 if not is_test: for m in CARD_RE.finditer(text): digits = re.sub(r"[ -]", "", m.group(0)) if 13 <= len(digits) <= 19 and digits[0] in "3456" and digits not in TEST_CARDS and luhn_ok(digits) \ and len(set(digits)) > 3: cards += 1 return emails, ids, cards # --------------------------------------------------------------------------------------------------------------- # Licences # --------------------------------------------------------------------------------------------------------------- LICENCE_FAMILY = { "MIT": "permissive", "Apache-2.0": "permissive", "BSD-2-Clause": "permissive", "BSD-3-Clause": "permissive", "ISC": "permissive", "Zlib": "permissive", "MPL-2.0": "weak_copyleft", "LGPL": "weak_copyleft", "EPL-2.0": "weak_copyleft", "GPL-2.0": "strong_copyleft", "GPL-3.0": "strong_copyleft", "GPL": "strong_copyleft", "AGPL-3.0": "network_copyleft", "Unlicense": "public_domain", "CC0-1.0": "public_domain", "CC-BY": "permissive", "BUSL-1.1": "source_available", "SSPL-1.0": "source_available", "proprietary": "proprietary", "unrecognized": "unknown", "none": "none", } def classify_licence_text(text): t = text[:20000] low = t.lower() if "gnu affero general public license" in low: return "AGPL-3.0" if "gnu lesser general public license" in low or "gnu library general public license" in low: return "LGPL" if "gnu general public license" in low: if "version 3" in low: return "GPL-3.0" if "version 2" in low: return "GPL-2.0" return "GPL" if "server side public license" in low: return "SSPL-1.0" if "business source license" in low: return "BUSL-1.1" if "mozilla public license" in low: return "MPL-2.0" if "eclipse public license" in low: return "EPL-2.0" if "apache license" in low and "2.0" in low: return "Apache-2.0" if "permission is hereby granted, free of charge" in low: return "MIT" if "redistribution and use in source and binary forms" in low: return "BSD-3-Clause" if "neither the name" in low or "endorse or promote" in low else "BSD-2-Clause" if "permission to use, copy, modify, and/or distribute this software" in low or low.strip().startswith("isc license"): return "ISC" if "this is free and unencumbered software released into the public domain" in low: return "Unlicense" if "cc0 1.0" in low or "creative commons zero" in low: return "CC0-1.0" if "creative commons" in low and "attribution" in low: return "CC-BY" if "zlib" in low and "altered source versions must be plainly marked" in low: return "Zlib" if any(w in low for w in ("all rights reserved", "proprietary", "confidential", "unauthorized copying", "not licensed", "no license is granted")): return "proprietary" return "unrecognized" def licence_from_spdx(raw): s = (raw or "").strip().strip("\"'") if not s: return None u = s.upper() if u in ("UNLICENSED", "PROPRIETARY", "COMMERCIAL", "PRIVATE") or u.startswith("SEE LICENSE") or u.startswith("LICENSEREF"): return "proprietary" for prefix, lic in (("AGPL", "AGPL-3.0"), ("LGPL", "LGPL"), ("GPL-2", "GPL-2.0"), ("GPL-3", "GPL-3.0"), ("GPL", "GPL"), ("MPL", "MPL-2.0"), ("EPL", "EPL-2.0"), ("APACHE", "Apache-2.0"), ("MIT", "MIT"), ("BSD-2", "BSD-2-Clause"), ("BSD-3", "BSD-3-Clause"), ("BSD", "BSD-3-Clause"), ("ISC", "ISC"), ("UNLICENSE", "Unlicense"), ("CC0", "CC0-1.0"), ("CC-BY", "CC-BY"), ("BUSL", "BUSL-1.1"), ("SSPL", "SSPL-1.0"), ("ZLIB", "Zlib"), ("0BSD", "ISC")): if u.startswith(prefix): return lic return "unrecognized" SPDX_HEADER_RE = re.compile(r"SPDX-License-Identifier:\s*([A-Za-z0-9.+\-() ]{2,80})") COPYRIGHT_RE = re.compile(r"(?i)copyright\s*(?:\(c\)|\u00a9)?\s*(?:\d{4}(?:\s*[-,]\s*\d{4})*,?)?\s*(?:by\s+)?([A-Za-z][^\n\r]{1,80})") HEADER_FAMILY_MARKERS = [ ("gnu affero general public", "network_copyleft"), ("gnu lesser general public", "weak_copyleft"), ("gnu general public license", "strong_copyleft"), ("mozilla public license", "weak_copyleft"), ("eclipse public license", "weak_copyleft"), ("licensed under the apache license", "permissive"), ("permission is hereby granted, free of charge", "permissive"), ("redistribution and use in source and binary forms", "permissive"), ] FAMILY_RANK = {"network_copyleft": 4, "strong_copyleft": 3, "weak_copyleft": 2, "source_available": 1, "permissive": 0, "public_domain": 0} def header_licence_family(head): best = None m = SPDX_HEADER_RE.search(head) if m: lic = licence_from_spdx(m.group(1).split(" OR ")[0].split(" AND ")[0]) fam = LICENCE_FAMILY.get(lic or "", "unknown") if fam in FAMILY_RANK: best = fam low = head.lower() for marker, fam in HEADER_FAMILY_MARKERS: if marker in low and (best is None or FAMILY_RANK[fam] > FAMILY_RANK[best]): best = fam return best, bool(m) def copyright_holder(head): m = COPYRIGHT_RE.search(head) if not m: return None holder = m.group(1) holder = re.split(r"(?i)all rights reserved|<|\(|licensed|\bsee\b|\.\s", holder)[0] holder = re.sub(r"[^a-z0-9 ]", "", holder.lower()).strip() holder = re.sub(r"\b(inc|llc|ltd|gmbh|corp|corporation|co|the|and|contributors|authors)\b", "", holder).strip() return holder or None # --------------------------------------------------------------------------------------------------------------- # Dependency manifests (parsed as text; nothing is installed or resolved) # --------------------------------------------------------------------------------------------------------------- LOCKFILES = { "package-lock.json": "npm", "npm-shrinkwrap.json": "npm", "pnpm-lock.yaml": "npm", "yarn.lock": "npm", "bun.lockb": "npm", "bun.lock": "npm", "poetry.lock": "pypi", "pipfile.lock": "pypi", "uv.lock": "pypi", "pdm.lock": "pypi", "cargo.lock": "cargo", "go.sum": "go", "gemfile.lock": "rubygems", "composer.lock": "composer", "packages.lock.json": "nuget", "gradle.lockfile": "gradle", "pubspec.lock": "pub", "mix.lock": "hex", } DEP_SIGNALS = { # dependency name (lower case) -> (framework or None, [domain tags]) "react": ("react", ["web_frontend"]), "react-dom": ("react", ["web_frontend"]), "next": ("nextjs", ["web_frontend", "web_backend"]), "vue": ("vue", ["web_frontend"]), "nuxt": ("nuxt", ["web_frontend"]), "@angular/core": ("angular", ["web_frontend"]), "svelte": ("svelte", ["web_frontend"]), "@sveltejs/kit": ("svelte", ["web_frontend", "web_backend"]), "express": ("express", ["web_backend"]), "@nestjs/core": ("nestjs", ["web_backend"]), "fastify": ("fastify", ["web_backend"]), "hono": (None, ["web_backend"]), "koa": (None, ["web_backend"]), "django": ("django", ["web_backend"]), "flask": ("flask", ["web_backend"]), "fastapi": ("fastapi", ["web_backend"]), "rails": ("rails", ["web_backend"]), "sinatra": (None, ["web_backend"]), "laravel/framework": ("laravel", ["web_backend"]), "symfony/framework-bundle": ("symfony", ["web_backend"]), "github.com/gin-gonic/gin": ("gin", ["web_backend"]), "github.com/labstack/echo/v4": ("echo", ["web_backend"]), "github.com/gofiber/fiber/v2": ("fiber", ["web_backend"]), "actix-web": ("actix", ["web_backend"]), "axum": ("axum", ["web_backend"]), "tokio": ("tokio", []), "torch": ("pytorch", ["machine_learning"]), "pytorch-lightning": ("pytorch", ["machine_learning"]), "tensorflow": ("tensorflow", ["machine_learning"]), "keras": ("tensorflow", ["machine_learning"]), "jax": ("jax", ["machine_learning"]), "scikit-learn": ("scikit_learn", ["machine_learning"]), "sklearn": ("scikit_learn", ["machine_learning"]), "xgboost": (None, ["machine_learning"]), "lightgbm": (None, ["machine_learning"]), "transformers": ("transformers", ["machine_learning", "llm_applications"]), "@tensorflow/tfjs": ("tensorflow", ["machine_learning"]), "onnxruntime": (None, ["machine_learning"]), "langchain": ("langchain", ["llm_applications"]), "llama-index": (None, ["llm_applications"]), "openai": (None, ["llm_applications"]), "anthropic": (None, ["llm_applications"]), "@anthropic-ai/sdk": (None, ["llm_applications"]), "pandas": ("pandas", ["data_engineering"]), "polars": (None, ["data_engineering"]), "pyspark": ("spark", ["data_engineering"]), "dask": (None, ["data_engineering"]), "apache-airflow": ("airflow", ["data_engineering"]), "dbt-core": ("dbt", ["data_engineering"]), "numpy": (None, ["scientific_computing"]), "scipy": (None, ["scientific_computing"]), "sympy": (None, ["scientific_computing"]), "opencv-python": (None, ["computer_vision"]), "opencv-python-headless": (None, ["computer_vision"]), "pillow": (None, []), "torchvision": (None, ["computer_vision", "machine_learning"]), "react-native": ("react_native", ["mobile"]), "expo": ("expo", ["mobile"]), "electron": ("electron", ["desktop"]), "graphql": ("graphql", ["web_backend"]), "@apollo/server": ("graphql", ["web_backend"]), "prisma": ("prisma", ["databases"]), "@prisma/client": ("prisma", ["databases"]), "sqlalchemy": ("sqlalchemy", ["databases"]), "typeorm": (None, ["databases"]), "sequelize": (None, ["databases"]), "drizzle-orm": (None, ["databases"]), "web3": (None, ["blockchain"]), "ethers": (None, ["blockchain"]), "hardhat": (None, ["blockchain"]), "@openzeppelin/contracts": (None, ["blockchain"]), "@solana/web3.js": (None, ["blockchain"]), "stripe": (None, ["fintech"]), "plaid": (None, ["fintech"]), "plaid-python": (None, ["fintech"]), "fhir.resources": (None, ["healthcare"]), "hl7": (None, ["healthcare"]), "fhirclient": (None, ["healthcare"]), "boto3": (None, ["cloud_infrastructure"]), "aws-sdk": (None, ["cloud_infrastructure"]), "@aws-sdk/client-s3": (None, ["cloud_infrastructure"]), "google-cloud-storage": (None, ["cloud_infrastructure"]), "azure-storage-blob": (None, ["cloud_infrastructure"]), "kubernetes": (None, ["devops"]), "pulumi": (None, ["cloud_infrastructure"]), "ansible": (None, ["devops"]), "pygame": (None, ["game_development"]), "phaser": (None, ["game_development"]), "three": (None, ["web_frontend"]), "cryptography": (None, ["security"]), "pycryptodome": (None, ["security"]), "commander": (None, ["developer_tools"]), "yargs": (None, ["developer_tools"]), "click": (None, ["developer_tools"]), "typer": (None, ["developer_tools"]), "github.com/spf13/cobra": (None, ["developer_tools"]), "clap": (None, ["developer_tools"]), } TEST_FRAMEWORK_DEPS = { "pytest": "pytest", "nose2": "nose2", "hypothesis": "hypothesis", "jest": "jest", "vitest": "vitest", "mocha": "mocha", "jasmine": "jasmine", "ava": "ava", "@playwright/test": "playwright", "playwright": "playwright", "cypress": "cypress", "karma": "karma", "rspec": "rspec", "rspec-rails": "rspec", "minitest": "minitest", "phpunit/phpunit": "phpunit", "pestphp/pest": "pest", "xunit": "xunit", "nunit": "nunit", "mstest.testframework": "mstest", "junit": "junit", "org.junit.jupiter": "junit", "testng": "testng", "io.kotest": "kotest", "spock-core": "spock", "scalatest": "scalatest", "test": "dart_test", "flutter_test": "flutter_test", "testthat": "testthat", "catch2": "catch2", "gtest": "googletest", "googletest": "googletest", } FRAMEWORK_RUNNERS = { "Python": {"pytest", "unittest", "nose2"}, "JavaScript": {"jest", "vitest", "mocha", "jasmine", "ava"}, "TypeScript": {"jest", "vitest", "mocha", "jasmine", "ava"}, "Go": {"go_test"}, "Rust": {"cargo_test"}, "Java": {"junit", "testng"}, "Kotlin": {"junit", "kotest"}, "Scala": {"scalatest", "junit"}, "Groovy": {"spock", "junit"}, "C#": {"xunit", "nunit", "mstest"}, "F#": {"xunit", "nunit"}, "Ruby": {"rspec", "minitest"}, "PHP": {"phpunit", "pest"}, "Swift": {"xctest"}, "Elixir": {"exunit"}, "Dart": {"dart_test", "flutter_test"}, "C": {"googletest", "ctest", "catch2"}, "C++": {"googletest", "ctest", "catch2"}, "R": {"testthat"}, "Vue": {"jest", "vitest"}, "Svelte": {"jest", "vitest"}, } DEFAULT_REGISTRIES = ("registry.npmjs.org", "registry.yarnpkg.com", "pypi.org", "files.pythonhosted.org", "repo.maven.apache.org", "repo1.maven.org", "api.nuget.org", "rubygems.org", "packagist.org", "crates.io", "proxy.golang.org", "pub.dev", "hex.pm", "plugins.gradle.org", "jcenter.bintray.com", "maven.google.com", "dl.google.com") class Deps: def __init__(self): self.components = [] # private: (ecosystem, name, spec, kind, scope, manifest path) self.manifests = Counter() self.direct = Counter() self.dev = Counter() self.lockfiles = set() self.git = 0 self.path = 0 self.url = 0 self.workspace = 0 self.escaping_path = 0 self.private_registry_configs = 0 self.private_scopes = set() self.private_scoped_packages = 0 self.frameworks = set() self.domains = set() self.test_frameworks = set() self.declared_licences = set() self.publish_hints = 0 self.workspace_tools = set() self.runtime_pins = set() self.test_command = False self.build_tools = set() self.own_names = set() # the project's own package/module names: only used to keep them out of the label def add(self, eco, name, spec, scope, manifest): name = (name or "").strip() if not name or len(name) > 200: return spec = (spec or "").strip() low = spec.lower() kind = "registry" if low.startswith(("git+", "git:", "git@", "github:", "gitlab:", "bitbucket:")) or "git=" in low or ".git" in low and "://" in low: kind = "git" elif low.startswith(("file:", "link:", "./", "../", "/")) or low.startswith("path="): kind = "path" elif low.startswith(("http://", "https://")): kind = "url" elif low.startswith("workspace:"): kind = "workspace" if kind == "git": self.git += 1 elif kind == "path": self.path += 1 target = re.sub(r"^(file:|link:|path=)", "", spec).strip("\"' ") base = os.path.dirname(manifest) joined = os.path.normpath(os.path.join(base, target)).replace("\\", "/") if joined.startswith("..") or os.path.isabs(target): self.escaping_path += 1 elif kind == "url": self.url += 1 elif kind == "workspace": self.workspace += 1 if scope == "dev": self.dev[eco] += 1 else: self.direct[eco] += 1 if name.startswith("@") and "/" in name and name.split("/")[0] in self.private_scopes: self.private_scoped_packages += 1 sig = DEP_SIGNALS.get(name.lower()) if sig: if sig[0]: self.frameworks.add(sig[0]) self.domains.update(sig[1]) tf = TEST_FRAMEWORK_DEPS.get(name.lower()) if tf and not (tf == "dart_test" and eco != "pub"): self.test_frameworks.add(tf) if eco in ("maven", "gradle") and "junit" in name.lower(): self.test_frameworks.add("junit") if eco in ("maven", "gradle") and "spring-boot" in name.lower(): self.frameworks.add("spring") self.domains.add("web_backend") if eco == "nuget": for k, v in (("xunit", "xunit"), ("nunit", "nunit"), ("mstest", "mstest")): if name.lower().startswith(k): self.test_frameworks.add(v) if name.lower().startswith("microsoft.aspnetcore"): self.frameworks.add("aspnet") self.domains.add("web_backend") self.components.append((eco, name, spec[:120], kind, scope, manifest)) def _toml_sections(text): """A deliberately small TOML reader: [section] -> list of raw 'key = value' lines (enough for manifests).""" sections, current = defaultdict(list), "" for raw in text.splitlines(): line = raw.split("#", 1)[0].rstrip() if not raw.strip().startswith("#") else "" if not line.strip(): continue m = re.match(r"^\s*\[\[?([^\]]+)\]\]?\s*$", line) if m: current = m.group(1).strip() continue sections[current].append(line.strip()) return sections def _toml_array(text, key): m = re.search(r"(?ms)^\s*" + re.escape(key) + r"\s*=\s*\[(.*?)\]", text) if not m: return [] return re.findall(r"[\"']([^\"']+)[\"']", m.group(1)) def _req_name(req): m = re.match(r"^\s*([A-Za-z0-9][A-Za-z0-9._\-\[\]]*)", req) return re.sub(r"\[.*\]$", "", m.group(1)).lower() if m else None OWN_NAME_RES = { "package.json": re.compile(r'"name"\s*:\s*"([^"\n]{2,200})"'), "composer.json": re.compile(r'"name"\s*:\s*"([^"\n]{2,200})"'), "pyproject.toml": re.compile(r"(?m)^\s*name\s*=\s*[\"']([^\"'\n]{2,200})[\"']"), "cargo.toml": re.compile(r"(?m)^\s*name\s*=\s*[\"']([^\"'\n]{2,200})[\"']"), "setup.cfg": re.compile(r"(?m)^\s*name\s*=\s*(\S{2,200})"), "go.mod": re.compile(r"(?m)^\s*module\s+(\S{2,200})"), } def parse_manifest(path, text, deps): name = path_parts(path)[-1] lower = name.lower() rx = OWN_NAME_RES.get(lower) if rx: for found in rx.findall(text[:200000])[:5]: deps.own_names.add(found.strip()) deps.own_names.update(p for p in re.split(r"[@/:]", found) if len(p) >= 4) try: if lower == "package.json": data = json.loads(text) if not isinstance(data, dict): return deps.manifests["npm"] += 1 for section, scope in (("dependencies", "prod"), ("devDependencies", "dev"), ("peerDependencies", "prod"), ("optionalDependencies", "prod")): sec = data.get(section) if isinstance(sec, dict): for k, v in sec.items(): deps.add("npm", k, v if isinstance(v, str) else "", scope, path) lic = data.get("license") if isinstance(lic, str): l2 = licence_from_spdx(lic) if l2: deps.declared_licences.add(l2) if data.get("name") and data.get("version") and data.get("private") is not True: deps.publish_hints += 1 if data.get("workspaces"): deps.workspace_tools.add("npm_workspaces") scripts = data.get("scripts") if isinstance(data.get("scripts"), dict) else {} test_script = scripts.get("test") if isinstance(test_script, str) and "no test specified" not in test_script: deps.test_command = True if scripts: deps.build_tools.add("npm_scripts") if isinstance(data.get("engines"), dict) and data["engines"].get("node"): deps.runtime_pins.add("node") elif lower.startswith("requirements") and lower.endswith(".txt"): deps.manifests["pypi"] += 1 scope = "dev" if "dev" in lower or "test" in lower else "prod" for raw in text.splitlines(): line = raw.split(" #", 1)[0].strip() if not line or line.startswith("#"): continue if line.startswith(("--index-url", "-i ", "--extra-index-url")): url = line.split(None, 1)[1] if " " in line else "" if not any(r in url for r in DEFAULT_REGISTRIES): deps.private_registry_configs += 1 continue if line.startswith(("-r", "-c", "--")): continue if line.startswith("-e"): target = line[2:].strip() deps.add("pypi", _req_name(target.split("#egg=")[-1]) or "editable", target, scope, path) continue if " @ " in line: nm, spec = line.split(" @ ", 1) deps.add("pypi", _req_name(nm), spec, scope, path) continue deps.add("pypi", _req_name(line), line, scope, path) elif lower == "pyproject.toml": deps.manifests["pypi"] += 1 for req in _toml_array(text, "dependencies"): deps.add("pypi", _req_name(req), req, "prod", path) secs = _toml_sections(text) for sec_name, lines in secs.items(): if sec_name in ("tool.poetry.dependencies", "tool.poetry.dev-dependencies") or sec_name.startswith("tool.poetry.group."): scope = "prod" if sec_name == "tool.poetry.dependencies" else "dev" for ln in lines: m = re.match(r"^([A-Za-z0-9_.\-]+)\s*=\s*(.+)$", ln) if m and m.group(1).lower() != "python": v = m.group(2) spec = "git=" if "git" in v and "=" in v else ("path=" + (re.search(r"path\s*=\s*[\"']([^\"']+)", v).group(1) if re.search(r"path\s*=\s*[\"']([^\"']+)", v) else "") if "path" in v else v) deps.add("pypi", m.group(1), spec, scope, path) m = re.search(r"(?m)^\s*license\s*=\s*(?:\{\s*text\s*=\s*)?[\"']([^\"']+)[\"']", text) if m: lic = licence_from_spdx(m.group(1)) if lic: deps.declared_licences.add(lic) if "[project]" in text or "[tool.poetry]" in text: if "Private :: Do Not Upload" not in text and re.search(r"(?m)^\s*name\s*=", text): deps.publish_hints += 1 if "[tool.pytest" in text: deps.test_frameworks.add("pytest") deps.test_command = True if "[tool.poetry]" in text: deps.build_tools.add("poetry") if "[tool.uv" in text: deps.build_tools.add("uv") if re.search(r"(?m)^\s*requires-python\s*=", text): deps.runtime_pins.add("python") elif lower == "setup.py" or lower == "setup.cfg": deps.manifests["pypi"] += 1 if lower == "setup.py": deps.publish_hints += 1 for req in re.findall(r"[\"']([A-Za-z0-9][A-Za-z0-9._\-]*(?:\[[^\]]*\])?\s*(?:[<>=!~]=?[^\"']*)?)[\"']", text.split("install_requires", 1)[1][:4000] if "install_requires" in text else ""): deps.add("pypi", _req_name(req), req, "prod", path) elif lower == "pipfile": deps.manifests["pypi"] += 1 deps.build_tools.add("pipenv") secs = _toml_sections(text) for sec_name, scope in (("packages", "prod"), ("dev-packages", "dev")): for ln in secs.get(sec_name, []): m = re.match(r"^\"?([A-Za-z0-9_.\-]+)\"?\s*=\s*(.+)$", ln) if m: deps.add("pypi", m.group(1), "git=" if "git" in m.group(2) else m.group(2), scope, path) elif lower == "go.mod": deps.manifests["go"] += 1 deps.build_tools.add("go_modules") if re.search(r"(?m)^go\s+\d", text): deps.runtime_pins.add("go") for m in re.finditer(r"(?m)^\s*(?:require\s+)?([a-zA-Z0-9.\-]+\.[a-z]{2,}/[^\s]+)\s+(v[^\s]+)(\s*//\s*indirect)?", text): if not m.group(3): deps.add("go", m.group(1), m.group(2), "prod", path) for m in re.finditer(r"(?m)=>\s*(\.{1,2}/[^\s]+)", text): deps.add("go", "replace", m.group(1), "prod", path) elif lower == "cargo.toml": deps.manifests["cargo"] += 1 deps.build_tools.add("cargo") secs = _toml_sections(text) if "workspace" in secs: deps.workspace_tools.add("cargo_workspace") for sec_name, lines in secs.items(): scope = "dev" if sec_name.endswith("dev-dependencies") else "prod" if sec_name.endswith("dependencies"): for ln in lines: m = re.match(r"^([A-Za-z0-9_\-]+)\s*=\s*(.+)$", ln) if m: v = m.group(2) pm = re.search(r"path\s*=\s*[\"']([^\"']+)", v) spec = "git=" if "git" in v and "=" in v else ("path=" + pm.group(1) if pm else v) deps.add("cargo", m.group(1), spec, scope, path) if "package" in secs: pkg = "\n".join(secs["package"]) if not re.search(r"publish\s*=\s*false", pkg): deps.publish_hints += 1 m = re.search(r"license\s*=\s*[\"']([^\"']+)", pkg) if m: lic = licence_from_spdx(m.group(1)) if lic: deps.declared_licences.add(lic) deps.test_frameworks.add("cargo_test") deps.test_command = True elif lower == "gemfile": deps.manifests["rubygems"] += 1 deps.build_tools.add("bundler") for m in re.finditer(r"(?m)^\s*gem\s+[\"']([^\"']+)[\"']([^\n]*)", text): rest = m.group(2) spec = "git=" if re.search(r"\b(git|github):", rest) else ("path=" + re.search(r"path:\s*[\"']([^\"']+)", rest).group(1) if re.search(r"path:\s*[\"']([^\"']+)", rest) else rest.strip(", ")) deps.add("rubygems", m.group(1), spec, "prod", path) elif lower.endswith(".gemspec"): deps.manifests["rubygems"] += 1 deps.publish_hints += 1 elif lower == "pom.xml": deps.manifests["maven"] += 1 deps.build_tools.add("maven") for m in re.finditer(r"(?s)(.*?)", text): block = m.group(1) g = re.search(r"([^<]+)", block) a = re.search(r"([^<]+)", block) s = re.search(r"([^<]+)", block) if a: deps.add("maven", ((g.group(1) + ":") if g else "") + a.group(1), "", "dev" if s and s.group(1) == "test" else "prod", path) for m in re.finditer(r"(?s).*?([^<]+)", text): if not any(r in m.group(1) for r in DEFAULT_REGISTRIES): deps.private_registry_configs += 1 if "" in text: deps.publish_hints += 1 elif lower in ("build.gradle", "build.gradle.kts"): deps.manifests["gradle"] += 1 deps.build_tools.add("gradle") for m in re.finditer(r"(?m)^\s*(implementation|api|compileOnly|runtimeOnly|testImplementation|testRuntimeOnly|kapt|ksp)\s*\(?\s*[\"']([^\"']+)[\"']", text): deps.add("gradle", m.group(2).rsplit(":", 1)[0] if m.group(2).count(":") >= 2 else m.group(2), m.group(2), "dev" if m.group(1).startswith("test") else "prod", path) for m in re.finditer(r"url\s*[=(]?\s*(?:uri\()?\s*[\"']([^\"']+)[\"']", text): if not any(r in m.group(1) for r in DEFAULT_REGISTRIES): deps.private_registry_configs += 1 if "maven-publish" in text: deps.publish_hints += 1 elif lower in ("settings.gradle", "settings.gradle.kts"): if re.search(r"(?m)^\s*include\b", text): deps.workspace_tools.add("gradle_multiproject") elif lower == "composer.json": data = json.loads(text) if not isinstance(data, dict): return deps.manifests["composer"] += 1 deps.build_tools.add("composer") for section, scope in (("require", "prod"), ("require-dev", "dev")): sec = data.get(section) if isinstance(sec, dict): for k, v in sec.items(): if k != "php" and not k.startswith("ext-"): deps.add("composer", k, v if isinstance(v, str) else "", scope, path) for repo in data.get("repositories") or []: if isinstance(repo, dict) and repo.get("type") in ("vcs", "git", "path"): deps.private_registry_configs += 1 lic = data.get("license") if isinstance(lic, str): l2 = licence_from_spdx(lic) if l2: deps.declared_licences.add(l2) if data.get("type") == "library" and data.get("name"): deps.publish_hints += 1 elif lower.endswith((".csproj", ".fsproj", ".vbproj")): deps.manifests["nuget"] += 1 deps.build_tools.add("dotnet") for m in re.finditer(r"true" in text or "true" in text: deps.publish_hints += 1 elif lower == "nuget.config": for m in re.finditer(r"= BATCH_BYTES or len(batch) >= 2000: for item in self._batch(batch): yield item batch, total = [], 0 if batch: for item in self._batch(batch): yield item def _batch(self, batch): data = self.git.run(["cat-file", "--batch"], input_bytes=("\n".join(e.oid for e in batch) + "\n").encode()).stdout pos = 0 for e in batch: nl = data.find(b"\n", pos) if nl < 0: return header = data[pos:nl].split() pos = nl + 1 if len(header) < 3 or header[1] == b"missing": yield e, None continue size = int(header[2]) yield e, data[pos:pos + size] pos += size + 1 def _is_link_dir(path): try: st = os.lstat(path) except OSError: return True return stat.S_ISLNK(st.st_mode) or bool(getattr(st, "st_file_attributes", 0) & 0x400) # FILE_ATTRIBUTE_REPARSE_POINT class DirectorySource: """For a folder without git: walks the files, never follows symlinks, skips dependency and build folders.""" SKIP = {".git", "node_modules", ".venv", "venv", "__pycache__", ".tox", ".next", "dist", "build", "target", ".gradle", ".idea", ".vscode", ".mypy_cache", ".pytest_cache", ".cache", "bower_components"} def __init__(self, root): self.root = root def entries(self): out = [] for dirpath, dirnames, filenames in os.walk(self.root, followlinks=False): # Windows junctions and other reparse points are links too (os.walk follows junctions before Python 3.12). dirnames[:] = sorted(d for d in dirnames if d not in self.SKIP and not _is_link_dir(os.path.join(dirpath, d))) for fn in sorted(filenames): full = os.path.join(dirpath, fn) rel = os.path.relpath(full, self.root).replace(os.sep, "/") try: st = os.lstat(full) except OSError: continue if stat.S_ISLNK(st.st_mode): out.append(Entry(rel, 0, None, "symlink")) elif stat.S_ISREG(st.st_mode): out.append(Entry(rel, st.st_size, None, "file")) return out def read_many(self, entries): for e in entries: try: fd = os.open(os.path.join(self.root, e.path), os.O_RDONLY | O_BINARY | getattr(os, "O_NOFOLLOW", 0)) with os.fdopen(fd, "rb") as fh: yield e, fh.read(MAX_ANALYZE_BYTES + 1) except OSError: yield e, None # --------------------------------------------------------------------------------------------------------------- # The scan # --------------------------------------------------------------------------------------------------------------- def _records(stream, sep, chunk=1 << 20): """Split a git output stream into records without holding the whole history in memory.""" buf = b"" while True: data = stream.read(chunk) if not data: break buf += data parts = buf.split(sep) buf = parts.pop() for part in parts: yield part if buf: yield buf FIX_RE = re.compile(r"(?i)\b(fix|fixes|fixed|fixing|bugfix|bugfixes|hotfix|hotfixes|bug|bugs|regression|resolves?\s+#\d+|closes?\s+#\d+)\b") CONVENTIONAL_FIX_RE = re.compile(r"^(fix|bugfix|hotfix)(\([^)]*\))?!?:", re.I) PR_PATTERNS = [ re.compile(r"Merge pull request #(\d+)"), re.compile(r"\(#(\d+)\)\s*$", re.M), re.compile(r"See merge request [^\s!]*!(\d+)"), re.compile(r"Merged in [^\n]*\(pull request #(\d+)\)"), re.compile(r"Merged PR (\d+)"), re.compile(r"Pull request #(\d+):"), ] BOT_RE = re.compile(r"(?i)\[bot\]|\bbot\b|dependabot|renovate|github-actions|greenkeeper|snyk|mergify|semantic-release|" r"allcontributors|pre-commit-ci|codecov|imgbot|deepsource|sourcery|gitlab-bot|azure-pipelines|" r"jenkins|buildkite|circleci|travis") class Scanner: def __init__(self, repo, opts): self.repo = os.path.abspath(repo) self.opts = opts self.quick = bool(opts.quick) self.started = time.time() self.limitations = set(["tests_not_executed", "secret_scan_is_heuristic", "personal_data_scan_is_heuristic", "dependency_licences_not_resolved"]) self.sensitive = set() # strings that must never appear in the share file (leak guard) self.path_names = set() # folder and file names: the label may not repeat one self.private = {"files": {}, "secrets": [], "personal_data": [], "rl_candidates": [], "notes": []} if self.quick: self.limitations.add("quick_mode") # ---- entry point ------------------------------------------------------------------------------------------- def run(self): if not os.path.isdir(self.repo): raise ScanError("not a folder") git = Git(self.repo) self.git_info = {"present": False, "has_commits": False, "shallow": False, "partial_clone": False, "worktree_dirty": False, "detached_head": False, "branch_ahead_of_default": None, "remote_hosting": [], "upstream_remote_present": False, "root_commits": 0} rev = None inside = git_available() and git.out(["rev-parse", "--is-inside-work-tree"], check=False) top = git.out(["rev-parse", "--show-toplevel"], check=False) if inside else None if inside and top and os.path.realpath(top.strip()) == os.path.realpath(self.repo): self.git_info["present"] = True rev = self._git_metadata(git) elif inside and top: raise ScanError("this folder is inside a larger git repository; point the check at the repository root") else: self.limitations.update(["no_git_metadata", "working_tree_without_git"]) if rev: source = GitTreeSource(git, rev) else: source = DirectorySource(self.repo) files = self._scan_files(source) history = self._history(git, rev) if rev else None hist_secrets = self._history_secrets(git, rev) if rev else None return self._assemble(files, history, hist_secrets) # ---- git metadata ------------------------------------------------------------------------------------------ def _git_metadata(self, git): info = self.git_info # Before any object is read: in a partial clone a missing object would be fetched from the remote, which can # run a program named in the repository's own config. git 2.44+ is told not to fetch; older git cannot be. promisor = git.out(["config", "--get-regexp", r"^(extensions\.partialclone|remote\..*\.promisor)$"], check=False) if promisor and promisor.strip(): if not git_can_refuse_lazy_fetch(): raise ScanError("this is a partial clone, and git %d.%d cannot be stopped from downloading missing objects; " "update git to 2.44 or newer, or check a full clone (nothing was written)" % _GIT["version"]) info["partial_clone"] = True self.limitations.add("partial_clone") ref = self.opts.ref or "HEAD" head = git.out(["rev-parse", "--verify", "--quiet", ref + "^{commit}"], check=False) if not head: if self.opts.ref: raise ScanError("the --ref you gave does not name a commit in this repository") self.limitations.add("no_commits") return None head = head.strip() info["has_commits"] = True self.sensitive.add(head) self.sensitive.add(head[:12]) if self.opts.ref: self.limitations.add("scanned_ref_not_checked_out") info["shallow"] = (git.out(["rev-parse", "--is-shallow-repository"], check=False) or "").strip() == "true" if info["shallow"]: self.limitations.add("shallow_history") branch = git.out(["symbolic-ref", "-q", "--short", "HEAD"], check=False) info["detached_head"] = not bool(branch and branch.strip()) if branch and branch.strip(): if len(branch.strip()) >= 6: self.sensitive.add(branch.strip()) roots = (git.out(["rev-list", "--max-parents=0", head], check=False) or "").split() info["root_commits"] = len(roots) for r in roots: self.sensitive.update([r, r[:12], r[:7]]) # In a shallow clone the parentless commits are only the clone's boundary, not the repository's first commits: # no reference is derived from them (a later full-clone scan would not match anyway). self.roots = [] if info["shallow"] else sorted(roots) remotes = git.out(["config", "--get-regexp", r"^remote\..*\.url$"], check=False) or "" hosting = set() for line in remotes.splitlines(): parts = line.split(None, 1) if len(parts) != 2: continue name, url = parts[0][len("remote."):-len(".url")], parts[1].strip() self.sensitive.add(url) for piece in re.split(r"[/:@]", url): if len(piece) >= 6 and piece.lower() not in ("github.com", "gitlab.com", "https", "bitbucket.org"): self.sensitive.add(piece.replace(".git", "")) if name == "upstream": info["upstream_remote_present"] = True host = url.lower() if "github.com" in host or "github." in host: hosting.add("github") elif "gitlab" in host: hosting.add("gitlab") elif "bitbucket" in host: hosting.add("bitbucket") elif "dev.azure.com" in host or "visualstudio.com" in host: hosting.add("azure_devops") elif host.startswith(("/", ".", "file:")) or re.match(r"^[a-z]:\\", host): hosting.add("local_path") else: hosting.add("other_host") info["remote_hosting"] = sorted(hosting) default = git.out(["symbolic-ref", "-q", "--short", "refs/remotes/origin/HEAD"], check=False) if default and default.strip(): ahead = git.out(["rev-list", "--count", default.strip() + ".." + head], check=False) if ahead and ahead.strip().isdigit(): info["branch_ahead_of_default"] = int(ahead.strip()) # Not `git status`: refreshing the index can run repository-defined clean filters on changed files. if self._worktree_dirty(git): info["worktree_dirty"] = True self.limitations.add("worktree_dirty_not_scanned") self.head = head return head def _worktree_dirty(self, git): """Uncommitted changes, found without letting git run anything: the index's blob ids are compared with our own hashes of the working files (CRLF-normalised as a fallback), and untracked files are listed from the index.""" fmt = (git.out(["rev-parse", "--show-object-format"], check=False) or "sha1").strip() or "sha1" algo = hashlib.sha256 if fmt == "sha256" else hashlib.sha1 raw = git.run(["ls-files", "-s", "-z"], check=False).stdout or b"" budget = 256 * 1024 * 1024 for rec in raw.split(b"\0"): if not rec: continue head, _, path = rec.partition(b"\t") bits = head.split() if len(bits) < 3 or bits[0] in (b"160000", b"120000"): continue full = os.path.join(self.repo, path.decode("utf-8", "replace")) try: st = os.lstat(full) except OSError: return True if not stat.S_ISREG(st.st_mode): return True if st.st_size > 64 * 1024 * 1024 or budget <= 0: continue budget -= st.st_size try: with open(full, "rb") as fh: data = fh.read() except OSError: continue want = bits[1].decode() if algo(b"blob %d\0" % len(data) + data).hexdigest() == want: continue lf = data.replace(b"\r\n", b"\n") if algo(b"blob %d\0" % len(lf) + lf).hexdigest() != want: return True others = git.run(["ls-files", "--others", "--exclude-standard", "-z", "--directory", "--no-empty-directory"], check=False).stdout or b"" return bool(others.strip(b"\0")) # ---- files ------------------------------------------------------------------------------------------------- def _scan_files(self, source): entries = source.entries() limit = QUICK_ANALYZE_BYTES if self.quick else MAX_ANALYZE_BYTES f = { "files_total": 0, "code_files": 0, "test_files": 0, "code_loc": 0, "test_loc": 0, "comment_loc": 0, "generated_files": 0, "generated_loc": 0, "vendored_files": 0, "vendored_loc": 0, "duplicate_files": 0, "duplicate_loc": 0, "minified_files": 0, "binary_files": 0, "large_files_skipped": 0, "lfs_pointer_files": 0, "submodules": 0, "symlinks": 0, "non_code_text_files": 0, } langs = defaultdict(lambda: {"files": 0, "code_loc": 0, "test_loc": 0, "comment_loc": 0}) dup_seen = {} risk = Counter() current_secret_keys = {} emails = {"code": set(), "tests": set(), "docs": set()} licence = {"root_licence": "none", "nested_licence_files": 0, "spdx": Counter(), "copyleft_header_files": 0, "copyright_header_files": 0, "holders": Counter(), "holder_files": []} deps = Deps() ci, quality, build, pins, ws = set(), set(), set(), set(), set() ci_runs_tests = False readme_bytes = 0 env_example_keys = 0 compose_services = 0 ai_files = 0 test_frameworks = set() manifest_dirs = set() to_read = [] self.paths = [] for e in entries: self.paths.append(e.path) if len(e.path) >= 5 and ("/" in e.path or "." in e.path): self.sensitive.add(e.path) if len(self.path_names) < 50000: for part in path_parts(e.path): stem = part.rsplit(".", 1)[0] if "." in part[1:] else part if len(stem) >= 4: self.path_names.add(stem) if e.kind == "submodule": f["submodules"] += 1 self.limitations.add("submodules_not_scanned") continue if e.kind == "symlink": f["symlinks"] += 1 continue f["files_total"] += 1 p = e.path lower_name = path_parts(p)[-1].lower() parts_lower = [x.lower() for x in path_parts(p)] # file-name signals that need no content if env_file(p): risk["env_files"] += 1 self.private["secrets"].append({"path": p, "line": None, "rule": "committed_env_file", "severity": "high"}) if any(r.search(p) for r in KEY_FILE_RES): risk["private_key_files"] += 1 self.private["secrets"].append({"path": p, "line": None, "rule": "key_or_certificate_file", "severity": "high"}) if any(r.search(p) for r in CREDENTIAL_FILE_RES): risk["credential_files"] += 1 ext = os.path.splitext(lower_name)[1] if ext in DATABASE_EXTENSIONS: risk["database_files"] += 1 if ext in DATA_EXTENSIONS and e.size > 1024 * 1024: risk["large_data_files"] += 1 if lower_name in AI_ASSISTANT_FILES or (len(parts_lower) >= 2 and parts_lower[-2] == "rules" and ".cursor" in parts_lower): ai_files += 1 self._config_signals(p, parts_lower, lower_name, ci, quality, build, pins, ws) if is_manifest(p) and not vendored_path(p): manifest_dirs.add(os.path.dirname(p)) if lower_name in LOCKFILES and not vendored_path(p): deps.lockfiles.add(LOCKFILES[lower_name]) if e.size > limit: f["large_files_skipped"] += 1 self.limitations.add("large_files_not_analyzed") continue to_read.append(e) for e, data in source.read_many(to_read): if data is None: continue p = e.path lower_name = path_parts(p)[-1].lower() head = data[:8000] if b"\0" in head: f["binary_files"] += 1 continue if data.startswith(b"version https://git-lfs.github.com/spec/"): f["lfs_pointer_files"] += 1 self.limitations.add("lfs_pointers_present") continue text = data.decode("utf-8", "replace") lang = language_of(p) test = is_test_path(p) vend = vendored_path(p) first = text[:3000] first_low = first.lower() gen = generated_path(p) or any(m in first_low[:2000] for m in GENERATED_MARKERS) # Manifests, CI and docs if is_manifest(p) and not vend: parse_manifest(p, text, deps) if self._is_ci_file(p) and re.search(r"(?i)\b(pytest|npm (run )?test|pnpm (run )?test|yarn test|go test|cargo test|mvn\b.*\btest|gradle\w* test|dotnet test|jest|vitest|rspec|phpunit|mix test|flutter test|ctest|tox|nox|make test|bundle exec rake)\b", text): ci_runs_tests = True if lower_name.startswith("readme") and len(path_parts(p)) == 1: readme_bytes = max(readme_bytes, len(data)) if env_example_file(p): env_example_keys += len(re.findall(r"(?m)^\s*(?:export\s+)?[A-Z][A-Z0-9_]{1,63}\s*=", text)) if lower_name in ("docker-compose.yml", "docker-compose.yaml", "compose.yml", "compose.yaml"): compose_services += self._compose_services(text) if lower_name in ("makefile", "gnumakefile") and re.search(r"(?m)^test\s*:", text): deps.test_command = True if lower_name in ("tox.ini", "noxfile.py", "pytest.ini") or (lower_name == "setup.cfg" and "[tool:pytest]" in text): deps.test_command = True if lower_name == "pytest.ini" or "[tool:pytest]" in text: test_frameworks.add("pytest") if lower_name in ("jest.config.js", "jest.config.ts", "jest.config.mjs", "jest.config.cjs"): test_frameworks.add("jest") if lower_name.startswith("vitest.config"): test_frameworks.add("vitest") if lower_name.startswith("playwright.config"): test_frameworks.add("playwright") if lower_name.startswith("cypress.config"): test_frameworks.add("cypress") # Licences if len(path_parts(p)) == 1 and re.match(r"^(licen[cs]e|copying|unlicense)(\.|$|-)", lower_name): licence["root_licence"] = classify_licence_text(text) elif re.match(r"^(licen[cs]e|copying|notice)(\.|$|-)", lower_name): licence["nested_licence_files"] += 1 # Secrets and personal data (all text files, including docs and config) in_docs = lang is None for rule, severity, line_no, key, documented in find_secrets(text): if documented: risk["documented_example_keys"] += 1 continue if test and rule == "generic_secret_assignment": risk["fixture_secret_strings_in_tests"] += 1 continue if test and rule in ("jwt", "connection_string_password"): risk["fixture_secret_strings_in_tests"] += 1 continue current_secret_keys[key] = (rule, severity) self.private["secrets"].append({"path": p, "line": line_no, "rule": rule, "severity": severity}) mails, ids, cards = find_personal_data(text, test) bucket = "tests" if test else ("docs" if in_docs else "code") emails[bucket].update(mails) if ids and not test: risk["national_id_like"] += ids risk["payment_card_like"] += cards if mails and not test: self.private["personal_data"].append({"path": p, "kind": "email_address", "count": len(mails)}) if ids and not test: self.private["personal_data"].append({"path": p, "kind": "national_id_like", "count": ids}) if cards: self.private["personal_data"].append({"path": p, "kind": "payment_card_like", "count": cards}) # Code accounting if lang is None: if non_code_kind(p): f["non_code_text_files"] += 1 continue code, comment = count_lines(text, lang) lines = text.count("\n") + 1 if lang in ("JavaScript", "CSS") and lines > 0 and (len(text) / lines > 300 or ".min." in lower_name): f["minified_files"] += 1 gen = True if vend: f["vendored_files"] += 1 f["vendored_loc"] += code continue if gen: f["generated_files"] += 1 f["generated_loc"] += code continue if code >= 10: h = normalized_hash(text) if h in dup_seen: f["duplicate_files"] += 1 f["duplicate_loc"] += code self.private["files"].setdefault("duplicates", []).append({"path": p, "same_as": dup_seen[h]}) continue dup_seen[h] = p head_text = "\n".join(text.splitlines()[:60]) fam, has_spdx = header_licence_family(head_text) if has_spdx or fam: licence["spdx"][fam or "other"] += 1 if fam in ("weak_copyleft", "strong_copyleft", "network_copyleft"): licence["copyleft_header_files"] += 1 self.private["files"].setdefault("copyleft_headers", []).append({"path": p, "family": fam}) holder = copyright_holder("\n".join(text.splitlines()[:40])) if holder: licence["copyright_header_files"] += 1 licence["holders"][holder] += 1 licence["holder_files"].append((p, holder)) if test: f["test_files"] += 1 f["test_loc"] += code langs[lang]["test_loc"] += code test_frameworks.update(self._frameworks_from_test(lang, text)) else: f["code_files"] += 1 f["code_loc"] += code langs[lang]["code_loc"] += code f["comment_loc"] += comment langs[lang]["files"] += 1 langs[lang]["comment_loc"] += comment if lang == "Go" and test: test_frameworks.add("go_test") if lang == "Rust" and "#[test]" in text: test_frameworks.add("cargo_test") if lang in ("C", "C++") and re.search(r"(?i)platformio|arduino|freertos|zephyr|stm32|esp32", first): deps.domains.add("embedded") if lang == "C#" and "using UnityEngine" in first: deps.domains.add("game_development") # ---- derived file facts ---- self.sensitive.update(h for h in licence["holders"] if h and len(h) >= 4) self.sensitive.update(n for n in deps.own_names if len(n) >= 6) # shorter pieces ("tool") only as whole labels self.path_names.update(deps.own_names) test_frameworks.update(deps.test_frameworks) monorepo = bool(ws or deps.workspace_tools) or len(manifest_dirs) >= 3 holders = licence["holders"] minority = 0 if holders: top_holder = holders.most_common(1)[0][0] for path, h in licence["holder_files"]: if h != top_holder: minority += 1 self.private["files"].setdefault("other_copyright_holders", []).append({"path": path}) self.current_secret_keys = current_secret_keys lang_list = [] for name, v in sorted(langs.items(), key=lambda kv: -(kv[1]["code_loc"] + kv[1]["test_loc"])): if v["code_loc"] + v["test_loc"] == 0 and v["files"] == 0: continue lang_list.append({"language": name, "files": v["files"], "code_loc": v["code_loc"], "test_loc": v["test_loc"], "comment_loc": v["comment_loc"]}) original = f["code_loc"] + f["test_loc"] rare = sum(v["code_loc"] + v["test_loc"] for k, v in langs.items() if k in RARE_STACKS) f["original_loc"] = original f["main_language"] = lang_list[0]["language"] if lang_list and original else None f["rare_stack_share_bp"] = bp(rare, original) f["test_to_code_bp"] = bp(f["test_loc"], f["code_loc"]) if f["code_loc"] else 0 if f["binary_files"]: self.limitations.add("binary_files_not_analyzed") self.langs_with_tests = {l["language"] for l in lang_list if l["test_loc"] > 0} self.test_frameworks = test_frameworks root_lic = licence["root_licence"] lockfile_ecos = deps.lockfiles ecosystems = [] for eco in ECOSYSTEMS: if deps.manifests[eco] or deps.direct[eco] or deps.dev[eco]: ecosystems.append({"ecosystem": eco, "manifests": deps.manifests[eco], "direct": deps.direct[eco], "dev": deps.dev[eco], "lockfile": eco in lockfile_ecos or (eco == "gradle" and "maven" in lockfile_ecos)}) with_lock = sum(1 for e in ecosystems if e["lockfile"]) build.update(deps.build_tools) pins.update(deps.runtime_pins) self.deps = deps readme_bucket = "none" if not readme_bytes else ("small" if readme_bytes < 1500 else ("medium" if readme_bytes < 8000 else "large")) return { "code": f, "languages": lang_list, "engineering": { "ci_systems": sorted(ci), "ci_runs_tests": ci_runs_tests, "test_frameworks": sorted(test_frameworks), "test_command_declared": deps.test_command, "quality_tools": sorted(quality), "monorepo": monorepo, "workspace_tools": sorted(ws | deps.workspace_tools), "package_manifests": sum(deps.manifests.values()), "frameworks": sorted(deps.frameworks), "domain_tags": sorted(deps.domains), "ai_assistant_config_files": ai_files, "readme": readme_bucket, }, "dependencies": { "ecosystems": ecosystems, "direct_total": sum(deps.direct.values()), "dev_total": sum(deps.dev.values()), "git_dependencies": deps.git, "path_dependencies": deps.path, "url_dependencies": deps.url, "workspace_dependencies": deps.workspace, "path_dependencies_outside_repo": deps.escaping_path, "private_registry_configs": deps.private_registry_configs, "private_scoped_packages": deps.private_scoped_packages, }, "licensing": { "root_licence": root_lic, "root_licence_family": LICENCE_FAMILY.get(root_lic, "unknown"), "nested_licence_files": licence["nested_licence_files"], "licence_header_files": {k: licence["spdx"].get(k, 0) for k in ("permissive", "weak_copyleft", "strong_copyleft", "network_copyleft", "source_available", "other")}, "copyleft_header_files": licence["copyleft_header_files"], "manifest_declared_licences": sorted(deps.declared_licences), "copyright_header_files": licence["copyright_header_files"], "copyright_holders_distinct": len(holders), "minority_copyright_holder_files": minority, "publish_hints": deps.publish_hints, }, "risk": risk, "emails": {k: len(v) for k, v in emails.items()}, "reproducibility": { "build_tools": sorted(build), "runtime_pins": sorted(pins), "containerized": "docker" in build, "devcontainer": "devcontainer" in build, "env_example_files": sum(1 for p in self.paths if env_example_file(p)), "env_example_keys": env_example_keys, "compose_services": compose_services, "lockfile_coverage_bp": bp(with_lock, len(ecosystems)) if ecosystems else 0, "build_blockers": deps.git + deps.escaping_path + deps.private_registry_configs + deps.private_scoped_packages, }, } def _is_ci_file(self, p): low = p.lower() return low.startswith(".github/workflows/") or low in (".gitlab-ci.yml", ".travis.yml", "azure-pipelines.yml", "bitbucket-pipelines.yml", "jenkinsfile", ".drone.yml", "appveyor.yml", "cloudbuild.yaml", "buildspec.yml") \ or low.startswith((".circleci/", ".buildkite/", ".woodpecker", ".semaphore/", ".teamcity/")) def _config_signals(self, p, parts, name, ci, quality, build, pins, ws): low = "/".join(parts) if low.startswith(".github/workflows/") and name.endswith((".yml", ".yaml")): ci.add("github_actions") mapping = {".gitlab-ci.yml": "gitlab_ci", ".travis.yml": "travis_ci", "azure-pipelines.yml": "azure_pipelines", "bitbucket-pipelines.yml": "bitbucket_pipelines", ".drone.yml": "drone", "appveyor.yml": "appveyor", ".appveyor.yml": "appveyor", "cloudbuild.yaml": "google_cloud_build", "buildspec.yml": "aws_codebuild"} if low in mapping: ci.add(mapping[low]) if name == "jenkinsfile": ci.add("jenkins") for prefix, system in ((".circleci/", "circleci"), (".buildkite/", "buildkite"), (".teamcity/", "teamcity"), (".woodpecker", "woodpecker"), (".semaphore/", "semaphore")): if low.startswith(prefix): ci.add(system) if name in (".eslintrc", ".eslintrc.js", ".eslintrc.json", ".eslintrc.cjs", ".eslintrc.yml", "eslint.config.js", "eslint.config.mjs", "eslint.config.ts", ".flake8", "ruff.toml", ".ruff.toml", ".pylintrc", ".rubocop.yml", ".golangci.yml", ".golangci.yaml", "clippy.toml", "checkstyle.xml", ".stylelintrc", "biome.json", "detekt.yml", ".swiftlint.yml", "phpcs.xml", ".credo.exs", "analysis_options.yaml"): quality.add("linter") if name in (".prettierrc", ".prettierrc.json", ".prettierrc.js", "prettier.config.js", ".clang-format", "rustfmt.toml", ".rustfmt.toml", ".editorconfig", ".php-cs-fixer.php", ".scalafmt.conf", ".formatter.exs") \ or name == "biome.json": quality.add("formatter") if name in ("tsconfig.json", "mypy.ini", ".mypy.ini", "pyrightconfig.json", "phpstan.neon", "psalm.xml", "flow-config"): quality.add("type_checker") if name in (".pre-commit-config.yaml", "lefthook.yml") or ".husky" in parts: quality.add("pre_commit") if name in (".coveragerc", "codecov.yml", ".codecov.yml", "coverage.xml", ".nycrc", ".nycrc.json", "jacoco.xml"): quality.add("coverage_config") if name == "codeowners": quality.add("codeowners") if name.startswith("contributing"): quality.add("contributing_guide") if name.startswith("security") and name.endswith((".md", ".txt", ".rst")): quality.add("security_policy") if name.startswith(("changelog", "changes", "history")) and name.endswith((".md", ".txt", ".rst")): quality.add("changelog") if parts and parts[0] in ("docs", "doc", "documentation") and len(parts) > 1: quality.add("docs_directory") if name in ("dependabot.yml", "renovate.json", ".renovaterc", "renovate.json5"): quality.add("dependency_updates") if name == "dockerfile" or name.startswith("dockerfile") or name.endswith(".dockerfile"): build.add("docker") if name in ("docker-compose.yml", "docker-compose.yaml", "compose.yml", "compose.yaml"): build.add("docker_compose") if ".devcontainer" in parts or name == ".devcontainer.json": build.add("devcontainer") if name in ("flake.nix", "shell.nix", "default.nix"): build.add("nix") if name in ("makefile", "gnumakefile"): build.add("makefile") if name == "justfile": build.add("justfile") if name in ("taskfile.yml", "taskfile.yaml"): build.add("taskfile") if name in ("workspace", "workspace.bazel", "module.bazel", "build.bazel"): build.add("bazel") ws.add("bazel") if name == "cmakelists.txt": build.add("cmake") if name == "tox.ini": build.add("tox") if name == "noxfile.py": build.add("nox") if name == "uv.lock": build.add("uv") if name == "poetry.lock": build.add("poetry") if name in ("environment.yml", "environment.yaml"): build.add("conda") if name == "build.sbt": build.add("sbt") if name in (".nvmrc", ".node-version"): pins.add("node") if name == ".python-version" or name == "runtime.txt": pins.add("python") if name == ".ruby-version": pins.add("ruby") if name in ("rust-toolchain", "rust-toolchain.toml"): pins.add("rust") if name == "global.json": pins.add("dotnet") if name in (".java-version", ".sdkmanrc"): pins.add("java") if name in (".tool-versions", "mise.toml", ".mise.toml"): pins.add("multi_tool") if name == "pnpm-workspace.yaml": ws.add("pnpm_workspace") if name == "lerna.json": ws.add("lerna") if name == "nx.json": ws.add("nx") if name == "turbo.json": ws.add("turbo") if name == "rush.json": ws.add("rush") if name == "go.work": ws.add("go_work") def _compose_services(self, text): count, inside = 0, False for line in text.splitlines(): if re.match(r"^services:\s*$", line): inside = True continue if inside and re.match(r"^\S", line): inside = False if inside and re.match(r"^ [A-Za-z0-9_.\-]+:\s*$", line): count += 1 return count def _frameworks_from_test(self, lang, text): found = set() head = text[:6000] if lang == "Python": if re.search(r"(?m)^\s*(import pytest|from pytest)", head) or "@pytest." in head: found.add("pytest") if re.search(r"(?m)^\s*(import unittest|from unittest)", head): found.add("unittest") if "from hypothesis" in head: found.add("hypothesis") elif lang in ("JavaScript", "TypeScript"): if "from \"vitest\"" in head or "from 'vitest'" in head: found.add("vitest") elif "@jest/globals" in head or "jest." in head: found.add("jest") if "@playwright/test" in head: found.add("playwright") elif lang in ("Java", "Kotlin"): if "org.junit" in head: found.add("junit") if "org.testng" in head: found.add("testng") if "io.kotest" in head: found.add("kotest") elif lang == "C#": for k, v in (("using Xunit", "xunit"), ("using NUnit", "nunit"), ("Microsoft.VisualStudio.TestTools", "mstest")): if k in head: found.add(v) elif lang == "Ruby": if "RSpec" in head or "describe " in head and "_spec.rb" in head: found.add("rspec") if "Minitest" in head: found.add("minitest") elif lang == "PHP" and "PHPUnit" in head: found.add("phpunit") elif lang == "Swift" and "import XCTest" in head: found.add("xctest") elif lang == "Elixir" and "ExUnit" in head: found.add("exunit") elif lang in ("C", "C++"): if "gtest/gtest.h" in head: found.add("googletest") if "catch2" in head.lower(): found.add("catch2") elif lang == "Dart" and "package:test/" in head: found.add("dart_test") elif lang == "Scala" and "org.scalatest" in head: found.add("scalatest") elif lang == "R" and "testthat" in head: found.add("testthat") return found # ---- history ----------------------------------------------------------------------------------------------- def _history(self, git, rev): h = {"commits": 0, "non_merge_commits": 0, "merge_commits": 0, "analyzed_commits": 0} total = git.out(["rev-list", "--count", rev], check=False) merges = git.out(["rev-list", "--count", "--merges", rev], check=False) h["commits"] = int(total.strip()) if total and total.strip().isdigit() else 0 h["merge_commits"] = int(merges.strip()) if merges and merges.strip().isdigit() else 0 h["non_merge_commits"] = max(0, h["commits"] - h["merge_commits"]) cap = QUICK_HISTORY_COMMITS if self.quick else MAX_HISTORY_COMMITS fmt = "%x1e%x1eH%x1f%H%x1f%P%x1f%aN%x1f%aE%x1f%at%x1f%ct%x1f%s%x1f%b%x1d" args = ["log", "--no-color", "--no-textconv", "--no-ext-diff", "--no-renames", "--numstat", "--summary", "--format=" + fmt, "--max-count=%d" % cap, rev, "--"] p = git.stream(args) records = _records(p.stdout, b"\x1e\x1eH\x1f") now = time.time() authors = {} # identity key -> canonical id alias = {} commit_author = [] months = Counter() years = Counter() day_counts = Counter() fix = fix_with_test = fix_adds_tests = feature_with_test = 0 prs = set() pr_commits = 0 post_ai = 0 added_total = added_post_ai = 0 anomalies = 0 first_ts = last_ts = None root_added = None candidates = [] cand_size = Counter() cand_lang = Counter() cand_runner = 0 bots = set() human_last12 = set() per_author = Counter() roots = set(getattr(self, "roots", [])) def ident(name, email): keys = [] if email: keys.append("e:" + email.strip().lower()) if name and len(name.strip()) > 2: keys.append("n:" + re.sub(r"\s+", " ", name.strip().lower())) found = next((alias[k] for k in keys if k in alias), None) cid = found or ("a%d" % len(authors)) authors[cid] = True for k in keys: alias.setdefault(k, cid) return cid for rec in records: if not rec.strip(): continue # The numstat block follows the record's last \x1d (a commit message may contain the byte itself; git # quotes control characters in paths). A message may also imitate a record start: its "commit" is skipped # unless it parses like one, and nothing below trusts a field's format. head, _, body = rec.rpartition(b"\x1d") fields = head.decode("utf-8", "replace").split("\x1f") if len(fields) < 8 or not re.match(r"^[0-9a-f]{40}([0-9a-f]{24})?$", fields[0]): continue sha, parents, name, email, at, ct, subject = fields[:7] message = "\x1f".join(fields[7:]) for s in (name, email): if s and len(s.strip()) >= 3: self.sensitive.add(s.strip()) self.sensitive.add(sha) if subject and len(subject.strip()) >= 12: self.sensitive.add(subject.strip()[:60]) h["analyzed_commits"] += 1 try: ts = clamp_ts(at) cts = clamp_ts(ct) except ValueError: continue is_merge = len(parents.split()) > 1 full = subject + "\n" + message for rx in PR_PATTERNS: nums = rx.findall(full) if nums: prs.update(nums) pr_commits += 1 break if ts > now + 86400 or ts < GIT_EPOCH or cts + 86400 < ts: anomalies += 1 if is_merge: continue is_bot = bool(BOT_RE.search(name or "") or BOT_RE.search((email or "").split("@")[0])) cid = ident(name, email) if is_bot: bots.add(cid) else: per_author[cid] += 1 if now - 365 * 86400 < ts <= now + 86400: human_last12.add(cid) commit_author.append((ts, cid, is_bot)) # A commit dated in the future is counted as a timestamp anomaly (above) and left out of the dates. if 0 < ts <= now + 86400: dt = utc_dt(ts) months[(dt.year, dt.month)] += 1 years[dt.year] += 1 day_counts[dt.date().isoformat()] += 1 first_ts = ts if first_ts is None or ts < first_ts else first_ts last_ts = ts if last_ts is None or ts > last_ts else last_ts if ts >= AI_ERA_START: post_ai += 1 # numstat / summary src_lines = test_lines = test_added = 0 src_files, test_files = 0, 0 lang_counter = Counter() added_here = 0 created_tests = 0 files_private = [] for line in body.decode("utf-8", "replace").splitlines(): if not line: continue if line.startswith(" create mode"): path = line.split(None, 3)[-1] if len(line.split(None, 3)) == 4 else "" if is_test_path(path) and language_of(path): created_tests += 1 continue if line.startswith(" "): continue parts = line.split("\t") if len(parts) != 3: continue a, d, path = parts if not (a.isdigit() and d.isdigit()): continue # binary changes ("-") and anything that is not a numstat line a, d = int(a), int(d) added_here += a lang = language_of(path) if not lang or vendored_path(path) or generated_path(path): continue files_private.append(path) if is_test_path(path): test_files += 1 test_lines += a + d test_added += a else: src_files += 1 src_lines += a + d lang_counter[lang] += a + d added_total += added_here if ts >= AI_ERA_START: added_post_ai += added_here if sha in roots: root_added = (root_added or 0) + added_here is_fix = bool(CONVENTIONAL_FIX_RE.match(subject) or FIX_RE.search(subject) or FIX_RE.search(message[:500])) if is_fix: fix += 1 if test_files and src_files: if is_fix: fix_with_test += 1 if test_added > 0 or created_tests: fix_adds_tests += 1 if src_lines <= 500 and src_files <= 15: size = "small" if src_lines <= 50 else ("medium" if src_lines <= 200 else "large") cand_size[size] += 1 main_lang = lang_counter.most_common(1)[0][0] if lang_counter else None if main_lang: cand_lang[main_lang] += 1 runner = bool(main_lang and FRAMEWORK_RUNNERS.get(main_lang, set()) & self.test_frameworks) cand_runner += 1 if runner else 0 if len(candidates) < 500: candidates.append({"commit": sha, "date": utc_dt(ts).date().isoformat(), "size": size, "language": main_lang, "source_lines_changed": src_lines, "test_lines_changed": test_lines, "test_runner_detected": runner, "files": files_private[:20]}) else: feature_with_test += 1 p.wait() if h["analyzed_commits"] < h["commits"]: self.limitations.add("history_sampled") human_total = sum(per_author.values()) core = sum(1 for c, n in per_author.items() if human_total and n >= max(3, 0.05 * human_total)) top_share = bp(per_author.most_common(1)[0][1], human_total) if per_author else 0 # activity clusters: split month sequence at gaps longer than 12 months, keep the busiest cluster ordered = sorted(months) clusters, cur = [], [] prev = None for ym in ordered: idx = ym[0] * 12 + ym[1] if prev is not None and idx - prev > 12: clusters.append(cur) cur = [] cur.append(ym) prev = idx if cur: clusters.append(cur) busiest = max(clusters, key=lambda c: sum(months[m] for m in c)) if clusters else [] active_span = ((busiest[-1][0] * 12 + busiest[-1][1]) - (busiest[0][0] * 12 + busiest[0][1]) + 1) if busiest else 0 self.private["rl_candidates"] = candidates hist = dict(h) hist.update({ "merged_prs_measured": len(prs) if prs else pr_commits if pr_commits else 0, "merged_prs_basis": "commit_messages", "human_contributors": len([c for c in per_author if c not in bots]), "bot_contributors": len(bots), "core_contributors": core, "top_contributor_share_bp": top_share, "contributors_active_last_12_months": len(human_last12), "first_commit_year": utc_dt(first_ts).year if first_ts else None, "last_commit_year": utc_dt(last_ts).year if last_ts else None, "history_span_months": calendar_months(first_ts, last_ts), "active_span_months": active_span, "active_months": len(months), "activity_clusters": len(clusters), "months_since_last_commit": months_between(last_ts, now) if last_ts else None, "commits_by_year": [{"year": y, "commits": years[y]} for y in sorted(years)], "fix_commits": fix, "fix_with_test_commits": fix_with_test, "fix_adding_test_lines_commits": fix_adds_tests, "feature_with_test_commits": feature_with_test, "post_2023_commit_share_bp": bp(post_ai, len(commit_author)), "post_2023_added_lines_share_bp": bp(added_post_ai, added_total), "timestamp_anomalies": anomalies, "max_single_day_commit_share_bp": bp(day_counts.most_common(1)[0][1], len(commit_author)) if day_counts else 0, "root_commit_added_lines_share_bp": bp(root_added, added_total) if root_added is not None and added_total else None, }) self.rl = {"total": sum(cand_size.values()), "small": cand_size["small"], "medium": cand_size["medium"], "large": cand_size["large"], "with_detected_test_runner": cand_runner, "languages": [{"language": k, "count": v} for k, v in cand_lang.most_common(8)]} return hist def _history_secrets(self, git, rev): if self.quick or self.opts.no_history_secrets: self.limitations.add("history_secret_scan_skipped") return None args = ["log", "-p", "--no-color", "--no-textconv", "--no-ext-diff", "--no-renames", "-U0", "--format=%x1eC%H", rev, "--", ".", ":(exclude,glob)**/node_modules/**", ":(exclude,glob)**/vendor/**", ":(exclude,glob)**/*.min.js", ":(exclude,glob)**/*-lock.json", ":(exclude,glob)**/*.lock"] p = git.stream(args) seen = {} total = 0 commit, path = None, None truncated = False buf = [] def flush(): if not buf or path is None: return text = "\n".join(buf) test = is_test_path(path) for rule, severity, _ln, key, documented in find_secrets(text): if documented or (test and rule in ("generic_secret_assignment", "jwt", "connection_string_password")): continue if key not in seen: seen[key] = (rule, severity, commit, path) for raw in p.stdout: total += len(raw) if total > MAX_HISTORY_PATCH_BYTES: truncated = True break line = raw.decode("utf-8", "replace").rstrip("\n") if line.startswith("\x1eC"): flush() buf = [] commit = line[2:] continue if line.startswith("+++ "): flush() buf = [] path = line[6:] if line.startswith("+++ b/") else None continue if line.startswith("+") and path: buf.append(line[1:]) if len(buf) > 5000: flush() buf = [] flush() if truncated: p.kill() self.limitations.add("history_secret_scan_truncated") p.wait() history_only = {k: v for k, v in seen.items() if k not in self.current_secret_keys} for k, (rule, severity, c, pth) in history_only.items(): self.private["secrets"].append({"path": pth, "line": None, "rule": rule, "severity": severity, "history_only": True, "commit": c[:12] if c else None}) kinds = sorted({v[0] for v in seen.values()} | {v[0] for v in self.current_secret_keys.values()}) return {"distinct": len(set(seen) | set(self.current_secret_keys)), "history_only": len(history_only), "kinds": kinds} # ---- assembly ---------------------------------------------------------------------------------------------- def _assemble(self, files, history, hist_secrets): risk = files["risk"] current = self.current_secret_keys sev = Counter(v[1] for v in current.values()) kinds = sorted({v[0] for v in current.values()}) if hist_secrets: kinds = hist_secrets["kinds"] rl = getattr(self, "rl", {"total": 0, "small": 0, "medium": 0, "large": 0, "with_detected_test_runner": 0, "languages": []}) code = files["code"] gi = self.git_info bulk = bool(history and not gi.get("shallow") and history.get("root_commit_added_lines_share_bp") is not None and history["root_commit_added_lines_share_bp"] >= 5000 and code["original_loc"] >= 5000) compressed = bool(history and history["commits"] < 20 and code["original_loc"] > 20000) declarations = { "source": "scanner_cli", "merged_prs": self.opts.declare_merged_prs, "never_public": self.opts.declare_never_public or "not_stated", "not_a_fork": self.opts.declare_not_fork or "not_stated", "owns_or_licensed": self.opts.declare_owner or "not_stated", "client_or_employer_consent": self.opts.declare_consent or "not_stated", "personal_data_present": self.opts.declare_personal_data or "not_stated", } measured = { "languages": files["languages"][:40], "code": {k: code[k] for k in ( "files_total", "code_files", "test_files", "code_loc", "test_loc", "comment_loc", "original_loc", "generated_files", "generated_loc", "vendored_files", "vendored_loc", "duplicate_files", "duplicate_loc", "minified_files", "binary_files", "large_files_skipped", "lfs_pointer_files", "submodules", "symlinks", "non_code_text_files", "main_language", "rare_stack_share_bp", "test_to_code_bp")}, "git": gi, "history": history, "engineering": files["engineering"], "dependencies": files["dependencies"], "licensing": files["licensing"], "risk_signals": { "secrets_current": len(current), "secrets_current_high": sev.get("high", 0), "secrets_current_medium": sev.get("medium", 0), "secrets_history_distinct": hist_secrets["distinct"] if hist_secrets else None, "secrets_history_only": hist_secrets["history_only"] if hist_secrets else None, "secret_kinds": kinds, "documented_example_keys": risk["documented_example_keys"], "fixture_secret_strings_in_tests": risk["fixture_secret_strings_in_tests"], "env_files": risk["env_files"], "private_key_files": risk["private_key_files"], "credential_files": risk["credential_files"], "emails_in_code": files["emails"]["code"], "emails_in_tests": files["emails"]["tests"], "emails_in_docs_and_config": files["emails"]["docs"], "national_id_like": risk["national_id_like"], "payment_card_like": risk["payment_card_like"], "database_files": risk["database_files"], "large_data_files": risk["large_data_files"], }, "reproducibility": files["reproducibility"], } heuristics = { "rl_task_candidates": dict(rl, method="fix-commit-with-test-change-v1", readiness="candidate_only"), "bulk_initial_import_suspected": bulk, "history_compressed_suspected": compressed, "fork_suspected": bool(gi.get("upstream_remote_present")), "public_distribution_suspected": files["licensing"]["publish_hints"] > 0 or files["licensing"]["root_licence_family"] in ("permissive", "weak_copyleft", "strong_copyleft", "network_copyleft", "public_domain"), "copied_code_suspected_files": code["duplicate_files"] + files["licensing"]["minority_copyright_holder_files"] + files["licensing"]["nested_licence_files"], } return {"measured": measured, "heuristics": heuristics, "declarations": declarations, "limitations": sorted(self.limitations), "duration_seconds": int(time.time() - self.started)} # --------------------------------------------------------------------------------------------------------------- # Report files # --------------------------------------------------------------------------------------------------------------- def local_key(opts): """A random key kept on this machine so repeat checks of the same repository produce the same reference, which nobody without the key (Monument included) can compute from a public copy of the repository.""" if opts.no_local_key: return None path = opts.key_file or os.path.join(os.path.expanduser("~"), ".monument-repo-check", "local-key") try: with open(path, "rb") as fh: key = fh.read().strip() if len(key) >= 32: return key except OSError: pass os.makedirs(os.path.dirname(path), exist_ok=True) key = _secrets.token_hex(32).encode() fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | O_BINARY, 0o600) with os.fdopen(fd, "wb") as fh: fh.write(key) return key def repository_ref(scanner, key): roots = getattr(scanner, "roots", None) if not roots: return None msg = ("monument-repo-ref/v1\0" + "\0".join(roots)).encode() if key is None: return {"method": REF_METHOD_EPHEMERAL, "value": hmac.new(_secrets.token_bytes(32), msg, hashlib.sha256).hexdigest()} return {"method": REF_METHOD_KEYED, "value": hmac.new(key, msg, hashlib.sha256).hexdigest()} def build_share(result, label, ref, report_id, generated_at): share = { "schema": SHARE_SCHEMA, "schema_version": SHARE_SCHEMA_VERSION, "report_id": report_id, "generated_at": generated_at, "scanner": {"name": SCANNER_NAME, "version": SCANNER_VERSION, "script_sha256": script_sha256(), "python": "%d.%d" % sys.version_info[:2], "platform": {"darwin": "macos", "linux": "linux", "win32": "windows"}.get(sys.platform, "other")}, "label": label, "repository_ref": ref, "scan": {"mode": "quick" if "quick_mode" in result["limitations"] else "full", "source": "working_tree" if "working_tree_without_git" in result["limitations"] else ("git_ref" if "scanned_ref_not_checked_out" in result["limitations"] else "git_head"), "duration_seconds": result["duration_seconds"], "limitations": result["limitations"]}, "measured": result["measured"], "heuristics": result["heuristics"], "declarations": result["declarations"], "verification": {"status": "unverified"}, } share["integrity"] = {"algorithm": INTEGRITY_ALGORITHM, "digest": hashlib.sha256(canonical_json(share).encode()).hexdigest()} return share VOCAB_VALUES = None def vocabulary(): return { "languages": LANGUAGES, "ci_systems": CI_SYSTEMS, "test_frameworks": TEST_FRAMEWORKS, "frameworks": FRAMEWORKS, "domain_tags": DOMAIN_TAGS, "quality_tools": QUALITY_TOOLS, "build_tools": BUILD_TOOLS, "runtime_pins": RUNTIME_PINS, "workspace_tools": WORKSPACE_TOOLS, "ecosystems": ECOSYSTEMS, "remote_hosting": REMOTE_HOSTING, "licences": LICENCES, "licence_families": LICENCE_FAMILIES, "secret_kinds": SECRET_KINDS, "limitations": LIMITATIONS, "declaration_answers": DECLARATION_ANSWERS, } ALLOWED_HEX_KEYS = {"script_sha256", "digest", "value", "report_id"} STRING_FIELDS_FREE = {"label", "generated_at", "python"} DEFAULT_LABEL_RE = re.compile(r"^Repository \d{1,4}$") def _label_norm(s): return re.sub(r"[\s._-]+", " ", (s or "").lower()).strip() def assert_share_is_safe(share, sensitive, names=()): """Refuse to write a share file that contains anything outside the documented vocabulary or any string seen in the repository (paths, names, emails, hashes, remotes, commit subjects). `names` are folder, file and package names: a label may not repeat one (or contain a long one).""" vocab = set() for values in vocabulary().values(): vocab.update(values) vocab.update({SHARE_SCHEMA, SHARE_SCHEMA_VERSION, SCANNER_NAME, SCANNER_VERSION, INTEGRITY_ALGORITHM, REF_METHOD_KEYED, REF_METHOD_EPHEMERAL, "quick", "full", "working_tree", "git_head", "git_ref", "scanner_cli", "unverified", "commit_messages", "fix-commit-with-test-change-v1", "candidate_only", "none", "small", "medium", "large", "macos", "linux", "windows", "other", "permissive", "weak_copyleft", "strong_copyleft", "network_copyleft", "source_available"}) def walk(node, key=None): if isinstance(node, dict): for k, v in node.items(): if not re.match(r"^[a-z0-9_]{1,64}$", k): raise LeakError("unexpected key in share report") walk(v, k) elif isinstance(node, list): for v in node: walk(v, key) elif isinstance(node, str): if key in ALLOWED_HEX_KEYS: if not re.match(r"^(rcr_)?[0-9a-f]{32,64}$", node): raise LeakError("unexpected identifier format") elif key in STRING_FIELDS_FREE: if key == "python" and not re.match(r"^\d\.\d{1,2}$", node): raise LeakError("unexpected interpreter version") if key == "label" and not LABEL_RE.match(node): raise LeakError("the label may use only letters, digits, spaces, dots, dashes and underscores") if key == "generated_at" and not re.match(r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z$", node): raise LeakError("unexpected timestamp") elif node not in vocab: raise LeakError("a value outside the fixed vocabulary was about to be written (%s)" % key) elif isinstance(node, bool) or node is None: return elif isinstance(node, int): if node < 0 or node > 10 ** 12: raise LeakError("number out of range") else: raise LeakError("only integers, booleans, null and fixed strings are allowed") walk(share) text = canonical_json(share) if EMAIL_RE.search(text): raise LeakError("an email address was about to be written") if "://" in text or "\\\\" in text: raise LeakError("a URL or path was about to be written") # Every string except the label is a fixed vocabulary value or a random/derived hex id (checked above), so the # only place repository text could appear is the label. The default "Repository N" is the scanner's own text. if not DEFAULT_LABEL_RE.match(share.get("label") or ""): label = (share.get("label") or "").lower() refuse = ("the label contains text from the repository (a folder, file or package name, a person's name or " "email, a copyright holder, a hash or part of a remote URL); choose a neutral label such as 'Repository 1'") for s in sensitive: if s and len(s) >= 4 and s.lower() in label: raise LeakError(refuse) nl = _label_norm(label) for n in names: nn = _label_norm(n) if len(nn) >= 4 and (nn == nl or (len(nn) >= 10 and nn in nl)): raise LeakError(refuse) for key in ("report_id",): if any(len(s) >= 7 and s.lower() in share[key] for s in sensitive if re.match(r"^[0-9a-f]{7,64}$", s or "")): raise LeakError("unexpected identifier") if len(text.encode()) > 64 * 1024: raise LeakError("share report exceeds 64 KB") def what_to_fix(share): m, items = share["measured"], [] r = m["risk_signals"] if r["secrets_current"]: items.append("Rotate and remove %d possible secret(s) in the current files (locations in this report)." % r["secrets_current"]) if r.get("secrets_history_only"): items.append("%d possible secret(s) exist only in git history: rotate them; removal needs a history rewrite you control." % r["secrets_history_only"]) if r["env_files"] or r["private_key_files"]: items.append("Remove committed .env or key files (%d) and rotate whatever they held." % (r["env_files"] + r["private_key_files"])) if r["emails_in_code"] or r["national_id_like"] or r["payment_card_like"]: items.append("Review possible personal data in code (emails %d, ID-like %d, card-like %d)." % (r["emails_in_code"], r["national_id_like"], r["payment_card_like"])) g = m["git"] if not g["present"]: items.append("Run the check on a git clone: history is a large part of what buyers evaluate.") elif g["shallow"]: items.append("This is a shallow clone: run the check on a full clone to measure the complete history.") if m["licensing"]["copyleft_header_files"]: items.append("Disclose or remove %d file(s) with copyleft licence headers." % m["licensing"]["copyleft_header_files"]) if m["licensing"]["root_licence"] in ("none", "unrecognized"): items.append("Add a licence or ownership statement so the rights position is explicit.") e = m["engineering"] if not e["ci_systems"]: items.append("No CI configuration found: buyers of training tasks need evidence that tests run.") if m["code"]["test_files"] == 0: items.append("No test files found: tests are required before anything can be considered for RL or evaluation tasks.") if m["dependencies"]["git_dependencies"] or m["dependencies"]["private_registry_configs"] or m["dependencies"]["path_dependencies_outside_repo"]: items.append("Build blockers: git, private-registry or out-of-repo path dependencies need a plan before a buyer can build the code.") if share["heuristics"]["public_distribution_suspected"]: items.append("Signals of public distribution (open-source licence or publishable manifest): confirm the code was never public.") return items def private_markdown(share, private, repo_path): m = share["measured"] lines = ["# Monument Repo Check: private report", "", "Keep this file. It contains file paths and locations. Send only `monument-share.json`, and only if you choose to.", "", "Repository folder: `%s`" % repo_path, "Label in the share file: `%s`" % share["label"], "Generated: %s, scanner %s" % (share["generated_at"], SCANNER_VERSION), ""] lines += ["## What to fix first", ""] fixes = what_to_fix(share) lines += ["- " + x for x in fixes] if fixes else ["- Nothing flagged by the automated checks. That is not proof that no secrets or personal data exist."] c, h = m["code"], m["history"] lines += ["", "## Size", "", "| Measure | Value |", "|---|---|", "| Original lines of code (code + tests, excluding generated, vendored, duplicates) | %d |" % c["original_loc"], "| Code lines / test lines / comment lines | %d / %d / %d |" % (c["code_loc"], c["test_loc"], c["comment_loc"]), "| Generated / vendored / duplicate lines left out | %d / %d / %d |" % (c["generated_loc"], c["vendored_loc"], c["duplicate_loc"]), "| Main language | %s |" % (c["main_language"] or "none")] if h: lines += ["", "## History", "", "| Measure | Value |", "|---|---|", "| Commits (merges) | %d (%d) |" % (h["commits"], h["merge_commits"]), "| Merged pull/merge requests found in messages | %d |" % h["merged_prs_measured"], "| Human contributors (core) | %d (%d) |" % (h["human_contributors"], h["core_contributors"]), "| History span / active span (months) | %d / %d |" % (h["history_span_months"], h["active_span_months"]), "| Fix commits / with test changes | %d / %d |" % (h["fix_commits"], h["fix_with_test_commits"])] lines += ["", "## Possible secrets (values are never shown)", "", "Automated rules find common formats only: this list is not proof that no other secrets exist.", ""] if private["secrets"]: lines += ["| Location | Rule | Severity | Only in history |", "|---|---|---|---|"] for s in private["secrets"][:300]: loc = s["path"] + (":%d" % s["line"] if s.get("line") else "") + ((" (commit %s)" % s["commit"]) if s.get("commit") else "") lines.append("| `%s` | %s | %s | %s |" % (loc, s["rule"], s["severity"], "yes" if s.get("history_only") else "no")) else: lines.append("None found by the automated rules.") lines += ["", "## Possible personal data", ""] if private["personal_data"]: lines += ["| File | Kind | Count |", "|---|---|---|"] + ["| `%s` | %s | %d |" % (p["path"], p["kind"], p["count"]) for p in private["personal_data"][:300]] else: lines.append("None found by the automated rules.") lines += ["", "## Candidate training-task commits (bug fixes that also changed tests)", "", "These are candidates only. A task is usable only after its tests are run and pass offline, under review.", ""] if private["rl_candidates"]: lines += ["| Commit | Date | Size | Language | Test runner detected |", "|---|---|---|---|---|"] for cnd in private["rl_candidates"][:100]: lines.append("| %s | %s | %s | %s | %s |" % (cnd["commit"][:12], cnd["date"], cnd["size"], cnd["language"] or "-", "yes" if cnd["test_runner_detected"] else "no")) else: lines.append("None found.") lines += ["", "## Evidence limitations", ""] + ["- " + x.replace("_", " ") for x in share["scan"]["limitations"]] lines += ["", "## Before you send anything", "", "Open `monument-share.json` and read it. It should contain only numbers, yes/no values and words from fixed lists,", "plus the label you chose. If anything in it looks like it identifies your code, people or company, do not send it.", ""] return "\n".join(lines) def dependency_csv(components): buf = io.StringIO() w = csv.writer(buf) w.writerow(["ecosystem", "name", "version_spec", "kind", "scope", "manifest"]) for row in components: w.writerow(row) return buf.getvalue() def scan_one(repo, out_dir, label, opts, key): repo_abs = os.path.abspath(repo) if is_within(out_dir, repo_abs): raise ScanError("the output folder is inside the repository; choose a folder outside it (nothing was written)") scanner = Scanner(repo_abs, opts) result = scanner.run() report_id = "rcr_" + _secrets.token_hex(16) generated_at = now_utc().strftime("%Y-%m-%dT%H:%M:%SZ") ref = repository_ref(scanner, key) share = build_share(result, label, ref, report_id, generated_at) sensitive = set(scanner.sensitive) sensitive.add(repo_abs) base = os.path.basename(repo_abs.rstrip(os.sep)) if len(base) >= 5: sensitive.add(base) # a shorter folder name is too generic to look for inside the label; it may not BE the label assert_share_is_safe(share, sensitive, scanner.path_names | {base}) os.makedirs(out_dir, exist_ok=True) private = { "schema": PRIVATE_SCHEMA, "schema_version": SHARE_SCHEMA_VERSION, "report_id": report_id, "generated_at": generated_at, "repository_folder": repo_abs, "share": share, "secrets": scanner.private["secrets"], "personal_data": scanner.private["personal_data"], "rl_task_candidates": scanner.private["rl_candidates"], "files": scanner.private["files"], "dependency_components": len(getattr(scanner, "deps", Deps()).components), "what_to_fix_first": what_to_fix(share), } write_file(os.path.join(out_dir, "monument-share.json"), json.dumps(share, indent=2, sort_keys=True) + "\n", private=False) write_file(os.path.join(out_dir, "monument-private-report.json"), json.dumps(private, indent=2, sort_keys=True) + "\n", private=True) write_file(os.path.join(out_dir, "monument-private-report.md"), private_markdown(share, scanner.private, repo_abs), private=True) write_file(os.path.join(out_dir, "monument-dependencies.csv"), dependency_csv(getattr(scanner, "deps", Deps()).components), private=True) return share def discover(parent): found = [] for name in sorted(os.listdir(parent)): full = os.path.join(parent, name) if os.path.isdir(full) and not os.path.islink(full) and os.path.exists(os.path.join(full, ".git")): found.append(full) return found def self_test(): """Builds a small synthetic repository in a temporary folder, checks it, and verifies the share file.""" tmp = tempfile.mkdtemp(prefix="monument-repo-check-selftest-") try: repo = os.path.join(tmp, "synthetic-repo") out = os.path.join(tmp, "out") os.makedirs(os.path.join(repo, "src")) os.makedirs(os.path.join(repo, "tests")) env = git_env() env.update({"GIT_AUTHOR_NAME": "Selftest Person", "GIT_AUTHOR_EMAIL": "selftest.person@corp-internal.invalid", "GIT_COMMITTER_NAME": "Selftest Person", "GIT_COMMITTER_EMAIL": "selftest.person@corp-internal.invalid"}) exe = resolve_git(avoid=[tmp]) if not exe: print("Self-test needs git on PATH.", file=sys.stderr) return 2 def g(*args): subprocess.run([exe, "-C", repo, "-c", "init.defaultBranch=main", "-c", "commit.gpgsign=false"] + list(args), check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, env=env) g("init", "-q") with open(os.path.join(repo, "src", "calc.py"), "w") as fh: fh.write("def add(a, b):\n return a + b\n") with open(os.path.join(repo, "tests", "test_calc.py"), "w") as fh: fh.write("import pytest\nfrom src.calc import add\n\ndef test_add():\n assert add(1, 2) == 3\n") g("add", "-A") g("commit", "-q", "-m", "initial calculator") with open(os.path.join(repo, "src", "calc.py"), "a") as fh: fh.write("\ndef sub(a, b):\n return a - b\n") with open(os.path.join(repo, "tests", "test_calc.py"), "a") as fh: fh.write("\ndef test_sub():\n assert sub(3, 1) == 2\n") g("add", "-A") g("commit", "-q", "-m", "fix: subtraction regression") opts = build_parser().parse_args(["--no-local-key", repo, "--out", out]) share = scan_one(repo, out, "Self Test", opts, None) text = open(os.path.join(out, "monument-share.json")).read() assert "selftest.person" not in text and "calc.py" not in text and "Selftest" not in text assert share["measured"]["history"]["fix_with_test_commits"] == 1 assert share["measured"]["code"]["test_files"] == 1 print("Self-test passed: the share file contained only counts and fixed labels.") return 0 finally: shutil.rmtree(tmp, ignore_errors=True) def build_parser(): p = argparse.ArgumentParser( prog="monument_repo_check.py", description="Local, read-only, offline repository check. Writes a shareable metadata report and a private detailed report.") p.add_argument("repos", nargs="*", help="one or more repository folders") p.add_argument("--out", help="output folder, outside every repository (one sub-folder per repository when checking several)") p.add_argument("--discover", metavar="FOLDER", help="check every git repository directly inside FOLDER") p.add_argument("--label", help="the only free text in the share file (letters, digits, spaces, . _ -; max 48)") p.add_argument("--quick", action="store_true", help="for very large repositories: samples history and skips the history secret scan") p.add_argument("--ref", help="check a branch, tag or commit instead of HEAD (read from git objects; nothing is checked out)") p.add_argument("--no-history-secrets", action="store_true", help="skip scanning git history for possible secrets") p.add_argument("--no-local-key", action="store_true", help="do not create or use ~/.monument-repo-check/local-key (repeat checks will not be linkable)") p.add_argument("--key-file", help=argparse.SUPPRESS) p.add_argument("--declare-merged-prs", type=int, metavar="N", help="your own count of merged pull/merge requests (recorded as your statement)") for flag, dest, extra in (("--declare-never-public", "declare_never_public", ""), ("--declare-not-fork", "declare_not_fork", ""), ("--declare-owner", "declare_owner", " (you own the code or hold the rights to license it)"), ("--declare-consent", "declare_consent", " (client or employer consent, where needed)"), ("--declare-personal-data", "declare_personal_data", " (the repository contains personal data)")): p.add_argument(flag, dest=dest, choices=["yes", "no"], help="your statement, recorded as a declaration%s" % extra) p.add_argument("--self-test", action="store_true", help="check a synthetic repository in a temporary folder and exit") p.add_argument("--print-vocabulary", action="store_true", help=argparse.SUPPRESS) p.add_argument("--version", action="version", version="%s %s" % (SCANNER_NAME, SCANNER_VERSION)) return p def main(argv=None): for stream in (sys.stdout, sys.stderr): try: stream.reconfigure(errors="replace") except (AttributeError, ValueError): pass opts = build_parser().parse_args(argv) if sys.version_info < (3, 8): print("Python 3.8 or newer is required.", file=sys.stderr) return 2 if opts.print_vocabulary: print(json.dumps(vocabulary(), indent=2, sort_keys=True)) return 0 if opts.self_test: return self_test() repos = list(opts.repos) if opts.discover: repos += discover(opts.discover) if not repos: build_parser().print_help() return 2 if not opts.out: print("Give an output folder outside the repository with --out.", file=sys.stderr) return 2 if opts.label and not LABEL_RE.match(opts.label): print("The label may use only letters, digits, spaces, dots, dashes and underscores (max 48).", file=sys.stderr) return 2 if opts.label and len(repos) > 1: print("--label applies to one repository; several repositories are labelled Repository 1, 2, ...", file=sys.stderr) return 2 if opts.declare_merged_prs is not None and not (0 <= opts.declare_merged_prs <= 10 ** 7): print("--declare-merged-prs must be between 0 and 10,000,000.", file=sys.stderr) return 2 out_root = os.path.abspath(os.path.expanduser(opts.out)) for r in repos: if is_within(out_root, os.path.abspath(r)): print("The output folder is inside a repository being checked. Choose a folder outside it. Nothing was written.", file=sys.stderr) return 2 if is_within(os.path.abspath(__file__), os.path.abspath(r)): print("This scanner file is inside a repository being checked. Save it outside every repository (for example in " "your Downloads folder) and run it from there with python3 -I. Nothing was written.", file=sys.stderr) return 2 resolve_git(avoid=[os.path.abspath(r) for r in repos] + ([os.path.abspath(opts.discover)] if opts.discover else [])) if not git_available(): print("Note: git was not found, so only file contents can be measured (no history).", file=sys.stderr) key = local_key(opts) shares, failures = [], 0 multi = len(repos) > 1 for i, repo in enumerate(repos, 1): label = opts.label or ("Repository %d" % i) out_dir = os.path.join(out_root, "repository-%d" % i) if multi else out_root try: share = scan_one(repo, out_dir, label, opts, key) shares.append(share) m = share["measured"] print("%s: %d original lines, %d test files, %s commits, %d possible secret(s) -> %s" % ( label, m["code"]["original_loc"], m["code"]["test_files"], m["history"]["commits"] if m["history"] else "no", m["risk_signals"]["secrets_current"], os.path.join(out_dir, "monument-share.json"))) except LeakError as e: failures += 1 print("%s: NOT written. The leak guard stopped the share file: %s" % (label, e), file=sys.stderr) except ScanError as e: failures += 1 print("%s: not checked: %s" % (label, e), file=sys.stderr) except Exception as e: # repository content the scanner did not expect: skip it, keep checking the others failures += 1 print("%s: not checked: unexpected %s while reading the repository (nothing was written for it)" % (label, type(e).__name__), file=sys.stderr) if multi: labels = "".join("Repository %d\t%s\n" % (i, os.path.abspath(r)) for i, r in enumerate(repos, 1)) os.makedirs(out_root, exist_ok=True) write_file(os.path.join(out_root, "labels-private.tsv"), labels, private=True) if shares: bundle = {"schema": BUNDLE_SCHEMA, "schema_version": SHARE_SCHEMA_VERSION, "reports": shares} write_file(os.path.join(out_root, "monument-share-bundle.json"), json.dumps(bundle, indent=2, sort_keys=True) + "\n", private=False) print("Bundle of %d share report(s): %s" % (len(shares), os.path.join(out_root, "monument-share-bundle.json"))) if shares: print("Read monument-share.json before sending it. Keep the private report files.") return 0 if failures == 0 else (3 if not shares else 1) if __name__ == "__main__": sys.exit(main())