#!/usr/bin/env python3 """ Monument sample checker — fully offline (Python 3.8+, standard library only). Checks a staged Monument sample folder, or the final ZIP itself, for credentials, personal and regulated data, record sets, linked identifiers, conversations, Git history, internal infrastructure, revealing file names and metadata, and for the Monument sample package format. It reports what it found WITHOUT printing any detected value. python3 monument_sample_check.py check-folder monument-sample-output python3 monument_sample_check.py check-zip monument-sample-output.zip python3 monument_sample_check.py hashes monument-sample-output (sizes and SHA-256 for the manifest and checksums) add --json for machine-readable output What it is: deterministic pattern rules, structural profiling and format checks that run on this computer. It opens no network connection, runs nothing it finds, uploads nothing and writes nothing. What it is not: proof that personal or confidential data is absent, proof of anonymization, or legal clearance. A clean result means only "no findings by these rules within what could be scanned". Images, PDFs, audio, video and other binary files cannot be checked and are never reported as clean. Exit codes: 0 CHECKS_PASSED (locally check-complete) · 1 NEEDS_CORRECTION · 10 REVIEW_REQUIRED · 20 SENSITIVE_DATA_FOUND · 30 UNSUPPORTED_CONTENT · 40 SCAN_INCOMPLETE · 50 SECURITY_HOLD · 2 usage error. The rules are the same file Monument's own checks use (config/client-samples/privacy-rules.json in the Monument repository); Monument repeats every check — and more — independently after secure submission. """ import argparse import base64 import calendar import datetime import getpass import hashlib import json import os import platform import re import shutil import stat import struct import subprocess import sys import tempfile import time import unicodedata import uuid import zipfile CHECKER_VERSION = "3.2.1" TOOL = "monument-sample-check" # The served checker carries the rules inline (replaced when Monument serves this file). RULES_JSON = r"""{ "rulesVersion": "client-sample-privacy-rules-2", "_comment": "Shared by the server checks (TypeScript), the browser Check My ZIP and the offline local checker (Python). Patterns use a portable regex subset (no lookbehind, no named groups); flags are given separately. A clean result means 'no findings by these rules within the scanned bytes' — never proof that personal or confidential data is absent.", "secrets": [ { "id": "private_key", "label": "private key", "pattern": "-----BEGIN (?:RSA |EC |DSA |OPENSSH |ENCRYPTED |PGP )?PRIVATE KEY(?: BLOCK)?-----" }, { "id": "aws_access_key_id", "label": "AWS access key id", "pattern": "\\b(?:AKIA|ASIA|AGPA|AIDA|AROA|ANPA|ANVA|AIPA)[A-Z0-9]{16}\\b" }, { "id": "aws_secret_key", "label": "AWS secret key", "pattern": "aws_?secret_?(?:access_?)?key[\"']?\\s*[:=]\\s*[\"']?[A-Za-z0-9/+]{40}", "flags": "i" }, { "id": "github_token", "label": "GitHub token", "pattern": "\\b(?:ghp|gho|ghu|ghs|ghr)_[A-Za-z0-9]{30,}\\b|\\bgithub_pat_[A-Za-z0-9_]{40,}\\b" }, { "id": "gitlab_token", "label": "GitLab token", "pattern": "\\bglpat-[A-Za-z0-9_-]{20,}\\b" }, { "id": "slack_token", "label": "Slack token", "pattern": "\\bxox[abposr]-[A-Za-z0-9-]{10,}\\b" }, { "id": "slack_webhook", "label": "Slack webhook", "pattern": "https://hooks\\.slack\\.com/services/[A-Za-z0-9/]{20,}" }, { "id": "stripe_key", "label": "Stripe key", "pattern": "\\b(?:sk|rk)_(?:live|test)_[A-Za-z0-9]{16,}\\b" }, { "id": "google_api_key", "label": "Google API key", "pattern": "\\bAIza[A-Za-z0-9_-]{35}\\b" }, { "id": "openai_key", "label": "OpenAI key", "pattern": "\\bsk-(?:proj-)?[A-Za-z0-9_-]{32,}\\b" }, { "id": "anthropic_key", "label": "Anthropic key", "pattern": "\\bsk-ant-[A-Za-z0-9_-]{20,}\\b" }, { "id": "npm_token", "label": "npm token", "pattern": "\\bnpm_[A-Za-z0-9]{30,}\\b|_authToken\\s*=\\s*\\S{10,}" }, { "id": "pypi_token", "label": "PyPI token", "pattern": "\\bpypi-[A-Za-z0-9_-]{40,}\\b" }, { "id": "sendgrid_key", "label": "SendGrid key", "pattern": "\\bSG\\.[A-Za-z0-9_-]{16,}\\.[A-Za-z0-9_-]{16,}\\b" }, { "id": "twilio_key", "label": "Twilio key", "pattern": "\\bSK[0-9a-f]{32}\\b" }, { "id": "digitalocean_token", "label": "DigitalOcean token", "pattern": "\\bdop_v1_[a-f0-9]{64}\\b" }, { "id": "shopify_token", "label": "Shopify token", "pattern": "\\bshp(?:at|ca|pa|ss)_[a-fA-F0-9]{32}\\b" }, { "id": "azure_storage_key", "label": "Azure storage key", "pattern": "AccountKey=[A-Za-z0-9+/=]{40,}" }, { "id": "jwt", "label": "JSON Web Token", "pattern": "\\beyJ[A-Za-z0-9_-]{10,}\\.eyJ[A-Za-z0-9_-]{10,}\\.[A-Za-z0-9_-]{10,}\\b" }, { "id": "url_with_credentials", "label": "URL with credentials", "pattern": "\\b[a-z][a-z0-9+.-]{1,20}://[^\\s:/@\"']{1,64}:[^\\s@/\"']{3,128}@[^\\s/\"']{2,}", "flags": "i" }, { "id": "authorization_header", "label": "authorization header", "pattern": "\\b(?:authorization|proxy-authorization)[\"']?\\s*[:=]\\s*[\"']?(?:bearer|basic|token)\\s+[A-Za-z0-9._~+/=-]{12,}", "flags": "i" }, { "id": "password_in_command", "label": "password on a command line", "pattern": "(?:^|\\s)-p[A-Za-z0-9!@#$%^&*._-]{6,}(?:\\s|$)|--password[= ]\\S{6,}" } ], "sensitiveKeys": { "_comment": "KEY=VALUE, key: value and \"key\": \"value\" with these key names and a non-placeholder value are credentials or sensitive configuration.", "pattern": "(?:^|[\\s{,\"'])([A-Za-z0-9_.-]{0,40}(?:password|passwd|pwd|secret|token|api[_-]?key|apikey|access[_-]?key|private[_-]?key|client[_-]?secret|auth[_-]?token|session[_-]?key|signing[_-]?key|encryption[_-]?key|database[_-]?url|db[_-]?url|dsn|connection[_-]?string|credentials?)[A-Za-z0-9_.-]{0,40})[\"']?\\s*[:=]\\s*[\"']?([^\\s\"'#,}]{6,})", "flags": "i", "placeholders": "^(?:<[^>]*>|\\$\\{[^}]*\\}|\\$[A-Z_][A-Z0-9_]*|%[A-Z_]+%|x{3,}|\\*{3,}|\\.{3,}|redacted|\\[redacted[^\\]]*\\]|changeme|change[_-]?me|example|placeholder|your[_-].*|none|null|true|false|undefined|todo|tbd|env\\(.*\\)|os\\.environ.*|process\\.env.*|getenv.*|secretkeyref.*|\\{\\{.*\\}\\})$" }, "pii": [ { "id": "email", "label": "email address", "kind": "direct", "pattern": "\\b[A-Za-z0-9._%+-]{1,64}@[A-Za-z0-9-]{1,63}(?:\\.[A-Za-z0-9-]{1,63})*\\.[A-Za-z]{2,24}\\b" }, { "id": "phone", "label": "phone number", "kind": "direct", "pattern": "(?:\\+\\d{1,3}[\\s.-]?)?(?:\\(\\d{2,5}\\)[\\s.-]?)?\\d{3,5}[\\s.-]\\d{3,4}[\\s.-]?\\d{3,4}\\b" }, { "id": "street_address", "label": "street address", "kind": "direct", "pattern": "\\b\\d{1,5}\\s+(?:[A-Z][a-z]+\\s){1,3}(?:Street|St|Road|Rd|Avenue|Ave|Lane|Ln|Drive|Dr|Boulevard|Blvd|Way|Close|Court|Ct|Place|Pl|Square|Sq|Terrace|Crescent|Highway|Hwy)\\b" }, { "id": "uk_postcode", "label": "UK postcode", "kind": "quasi", "pattern": "\\b[A-Z]{1,2}\\d[A-Z\\d]?\\s?\\d[A-Z]{2}\\b" }, { "id": "date_of_birth", "label": "date of birth", "kind": "quasi", "pattern": "\\b(?:dob|d\\.o\\.b\\.?|date[ _-]of[ _-]birth|birth[ _-]?date|born(?: on)?)\\b[\"']?\\s*[:=,]?\\s*[\"']?\\d{1,4}[-/.]\\d{1,2}[-/.]\\d{1,4}", "flags": "i" }, { "id": "ipv4", "label": "IP address", "kind": "quasi", "pattern": "\\b(?:(?:25[0-5]|2[0-4]\\d|1?\\d?\\d)\\.){3}(?:25[0-5]|2[0-4]\\d|1?\\d?\\d)\\b" } ], "regulated": [ { "id": "us_ssn", "label": "US social security number", "pattern": "\\b(?!000|666|9\\d\\d)\\d{3}-(?!00)\\d{2}-(?!0000)\\d{4}\\b" }, { "id": "uk_nino", "label": "UK National Insurance number", "pattern": "\\b[A-CEGHJ-PR-TW-Z][A-CEGHJ-NPR-TW-Z]\\s?\\d{2}\\s?\\d{2}\\s?\\d{2}\\s?[A-D]\\b" }, { "id": "iban", "label": "bank account (IBAN)", "pattern": "\\b[A-Z]{2}\\d{2}(?:\\s?[A-Z0-9]{4}){3,7}(?:\\s?[A-Z0-9]{1,4})?\\b" }, { "id": "payment_card", "label": "payment card number", "pattern": "\\b(?:\\d[ -]?){13,19}\\b", "luhn": true }, { "id": "bank_account", "label": "bank account details", "pattern": "\\b(?:sort[ _-]?code|routing(?: number)?|aba|account(?: number| no\\.?)?|acct)\\b[\"']?\\s*[:=#]?\\s*[\"']?\\d[\\d -]{5,}", "flags": "i" }, { "id": "medical_record", "label": "medical record number", "pattern": "\\b(?:mrn|medical record(?: number)?|patient(?: id| number| no\\.?)|nhs(?: number| no\\.?))\\b[\"']?\\s*[:=#]?\\s*[\"']?(?=[A-Z0-9-]*\\d)[A-Z0-9-]{5,}", "flags": "i" }, { "id": "government_document", "label": "passport / licence number", "pattern": "\\b(?:passport|driv(?:er'?s|ing) licen[cs]e|national id|tax id|tin|ein|vat number)\\b[\"']?\\s*(?:no\\.?|number|#)?\\s*[:=#]?\\s*[\"']?(?=[A-Z0-9-]*\\d)[A-Z0-9-]{6,}", "flags": "i" } ], "specialCategoryTerms": "\\b(?:diagnos(?:is|ed)|prescription|medication|patient|mrn|icd-?10|hiv|cancer|diabetes|pregnan(?:t|cy)|mental health|disabilit(?:y|ies)|therapy|blood type|genetic|ethnicity|ethnic origin|religio(?:n|us)|sexual orientation|trade union|criminal record|conviction|salary|payroll|bank statement|credit score)\\b", "personName": { "_comment": "Heuristics, not proof: honorific or label followed by capitalised words, greetings, and name-like values in person fields.", "patterns": [ "\\b(?:Mr|Mrs|Ms|Miss|Mx|Dr|Prof)\\.?\\s+[A-Z][a-z]+(?:\\s+[A-Z][a-z]+)?\\b", "\\b(?:[Cc]ustomer|[Cc]lient|[Pp]atient|[Ee]mployee|[Uu]ser|[Cc]aller|[Cc]ontact|[Aa]pplicant|[Tt]enant|[Mm]ember|[Aa]gent|[Ee]ngineer)\\s+[A-Z][a-z]+\\s+[A-Z][a-z]+\\b", "\\b(?:[Nn]ame|[Ff]ull name|[Cc]ontact name|[Ss]igned by|[Pp]repared by|[Aa]uthor)\\s*[:=]\\s*[A-Z][a-z]+\\s+[A-Z][a-z]+\\b", "\\b(?:Dear|Hi|Hello|Thanks|Thank you)\\s+[A-Z][a-z]+(?:\\s+[A-Z][a-z]+)?[,!]", "\\bI(?:'m| am)\\s+[A-Z][a-z]+\\s+[A-Z][a-z]+\\b" ], "valuePattern": "^[A-Z][a-z'-]+(?:\\s+[A-Z][a-z'-]+){1,3}$" }, "internalInfrastructure": [ { "id": "internal_hostname", "label": "internal hostname", "pattern": "\\b[a-z0-9-]{1,63}(?:\\.[a-z0-9-]{1,63}){0,8}\\.(?:internal|corp|intranet|lan|local|localdomain|svc\\.cluster\\.local)\\b", "flags": "i" }, { "id": "private_ip", "label": "private network address", "pattern": "\\b(?:10\\.(?:\\d{1,3}\\.){2}\\d{1,3}|192\\.168\\.\\d{1,3}\\.\\d{1,3}|172\\.(?:1[6-9]|2\\d|3[01])\\.\\d{1,3}\\.\\d{1,3})\\b" }, { "id": "cloud_account", "label": "cloud account identifier", "pattern": "\\barn:aws[a-z-]*:[a-z0-9-]+:[a-z0-9-]*:\\d{12}:|\\b\\d{12}\\.dkr\\.ecr\\.|projects/[a-z][a-z0-9-]{4,28}[a-z0-9]/(?:secrets|serviceAccounts)", "flags": "i" }, { "id": "internal_tool_url", "label": "internal tool URL", "pattern": "https?://[a-z0-9.-]+\\.(?:atlassian\\.net|slack\\.com/archives|sharepoint\\.com|zendesk\\.com/agent|salesforce\\.com|my\\.salesforce\\.com)\\b", "flags": "i" } ], "gitHistory": "^(?:diff --git |From [0-9a-f]{7,40} |commit [0-9a-f]{40}\\b|Author:\\s.+<[^>]+>|Signed-off-by:\\s|Co-authored-by:\\s|index [0-9a-f]{7,}\\.\\.[0-9a-f]{7,})", "conversationKeys": [ "messages", "conversation", "conversations", "transcript", "utterance", "utterances", "dialog", "dialogue", "turns", "chat", "chat_history", "history" ], "conversationRoleKeys": [ "role", "speaker", "author", "sender", "from" ], "conversationTextMarkers": "^(?:user|assistant|agent|customer|caller|human|ai|operator|support|system)\\s*:", "headers": { "_comment": "Column / key names (lower-case, separators removed) that mark a field as identifying.", "direct": [ "name", "firstname", "lastname", "fullname", "surname", "givenname", "familyname", "middlename", "customername", "clientname", "patientname", "employeename", "contactname", "username", "displayname", "email", "emailaddress", "mail", "phone", "phonenumber", "mobile", "telephone", "tel", "cell", "address", "streetaddress", "street", "addressline1", "addressline2", "ssn", "socialsecuritynumber", "nino", "nationalid", "passport", "passportnumber", "driverslicense", "iban", "accountnumber", "cardnumber", "creditcard", "mrn", "nhsnumber", "taxid" ], "quasi": [ "postcode", "postalcode", "zip", "zipcode", "dob", "dateofbirth", "birthdate", "birthday", "age", "gender", "sex", "city", "town", "county", "state", "region", "jobtitle", "title", "occupation", "employer", "company", "department", "ethnicity", "race", "nationality", "religion", "maritalstatus", "salary", "income", "diagnosis", "condition" ], "personId": [ "customerid", "userid", "accountid", "patientid", "employeeid", "memberid", "clientid", "personid", "contactid", "subscriberid", "studentid", "staffid", "uid", "guestid", "policyholderid" ] }, "sensitivePathTerms": "(?:^|[/_. -])(?:payroll|salar(?:y|ies)|hr|personnel|employees?|staff[_-]?list|customers?|clients?[_-]?list|patients?|medical|health|ssn|passports?|contracts?|nda|confidential|internal[_-]?only|secrets?|credentials?|passwords?|private|bank|tax|invoices?|crm[_-]?export|leads|users?[_-]?export|backup|dump)(?:[/_. -]|$)", "limits": { "maxSampleFiles": { "metadata_only": 0, "synthetic_or_authorized_examples": 50, "real_excerpts": 25 }, "maxSampleBytes": { "metadata_only": 0, "synthetic_or_authorized_examples": 5242880, "real_excerpts": 2097152 }, "maxRecordsPerTable": { "synthetic_or_authorized_examples": 500, "real_excerpts": 100 }, "recordsThreshold": 3, "linkageMinOccurrences": 2, "quasiIdentifierThreshold": 3, "encodedMinLength": 32, "encodedMaxDecodeBytes": 65536 } }""" # Monument's server limits for code samples (policy codeSamples.scanBytesTotal / scanBytesPerEntry), set by the server. CODE_SCAN_LIMITS_JSON = r"""{"scanBytesTotal":67108864,"scanBytesPerEntry":2097152,"maxEntries":2000}""" CONTRACT_VERSION = "3.0" REQUIRED_FILES = ["README.md", "ASSET_OVERVIEW.md", "DATA_DICTIONARY.md", "RIGHTS_AND_RESTRICTIONS.md", "PRIVACY_AND_REVIEW.md", "REVIEW_REQUIRED.md", "sample_manifest.json", "CHECKSUMS.sha256"] DOCUMENTS = [f for f in REQUIRED_FILES if f.endswith(".md")] LEGACY_V1_FILES = ["PROVENANCE.md", "SANITIZATION_REPORT.md", "LIMITATIONS.md"] SAMPLE_DIRS = {"synthetic": "SAMPLES/SYNTHETIC/", "authorized": "SAMPLES/AUTHORIZED/", "excerpts": "SAMPLES/EXCERPTS/"} # Limits (mirroring config/client-samples/policy.yaml → archive; the server's limits are authoritative). MAX_ENTRIES = 500 MAX_UNCOMPRESSED = 64 * 1024 * 1024 MAX_ENTRY_BYTES = 25 * 1024 * 1024 MAX_RATIO = 100 RATIO_FLOOR = 65536 MAX_MANIFEST_BYTES = 262144 MAX_DOCUMENT_BYTES = 2097152 SCAN_PER_ENTRY = 2097152 SCAN_TOTAL = 20971520 TEXT_EXT = re.compile(r"\.(md|markdown|txt|text|csv|tsv|json|jsonl|ndjson|ya?ml|toml|xml|html?|ini|cfg|conf|env|properties|log|sql|patch|diff|rst|adoc|tex|[a-z]{1,5}rc)$", re.I) SOURCE_EXT = re.compile(r"\.(js|mjs|cjs|jsx|ts|tsx|py|rb|go|rs|java|kt|kts|scala|c|h|cc|cpp|hpp|cs|swift|m|mm|php|pl|pm|r|jl|lua|sh|bash|zsh|ps1|psm1|bat|cmd|vbs|sql|ipynb|tf|groovy|dart|ex|exs|erl|hs|clj|vue|svelte)$", re.I) OOXML_EXT = re.compile(r"\.(docx|xlsx|pptx)$", re.I) OOXML_TEXT_PART = re.compile(r"^(word/(document|comments|footnotes|endnotes|header\d*|footer\d*)|xl/(sharedStrings|worksheets/sheet\d+)|ppt/(slides/slide\d+|notesSlides/notesSlide\d+))\.xml$", re.I) OOXML_META_PART = re.compile(r"^docProps/(core|app|custom)\.xml$", re.I) EXECUTABLE_EXT = re.compile(r"\.(exe|dll|scr|com|msi|msp|mst|jar|war|apk|ipa|app|lnk|url|scf|pif|iso|img|vhd|vhdx|dmg|pkg|deb|rpm|hta|cpl|ocx|sys|so|dylib|bin|elf|class|reg|inf|desktop|webloc|library-ms|searchconnector-ms|settingcontent-ms|appref-ms|application|gadget|msc|sct|chm|wsc|ws|wsf|wsh|vbe|jse|ps1xml|psc1|xbap|xll|iqy|slk)$", re.I) ARCHIVE_EXT = re.compile(r"\.(zip|tar|tgz|gz|bz2|xz|7z|rar|zst|lz|lzma|cab)$", re.I) DEPENDENCY_DIR = re.compile(r"(^|/)(node_modules|\.git|\.svn|\.hg|__pycache__|\.venv|venv|\.tox|vendor/bundle|target|dist|build)/") OS_METADATA_NAME = re.compile(r"(^|/)(__MACOSX(/|$)|\.DS_Store$|\._[^/]*$)", re.I) SECRET_FILE = re.compile(r"(^|/)(\.env(\..+)?|id_(rsa|dsa|ecdsa|ed25519)(\.pub)?|[^/]*\.(pem|key|p12|pfx|keystore|jks|kdbx|ppk)|credentials(\.json)?|\.npmrc|\.pypirc|\.netrc|\.htpasswd|\.pgpass|\.git-credentials|\.bash_history|\.zsh_history|\.psql_history|\.mysql_history|kubeconfig|service[-_]?account[^/]*\.json|secrets?\.(ya?ml|json|toml))$", re.I) SHA = re.compile(r"^[0-9a-f]{64}$") STATUS_EXIT = {"CHECKS_PASSED": 0, "NEEDS_CORRECTION": 1, "REVIEW_REQUIRED": 10, "SENSITIVE_DATA_FOUND": 20, "UNSUPPORTED_CONTENT": 30, "SCAN_INCOMPLETE": 40, "SECURITY_HOLD": 50} LIMITS_TEXT = [ "These checks use deterministic rules on this computer. They do not prove that personal or confidential data is absent, that anything is anonymized, or that anyone has the right to share it. A clean result is not legal clearance.", "Images, PDFs, audio, video and other binary files cannot be checked here and are never reported as clean.", "Synthetic examples show format only; they say nothing about the quality or characteristics of the real data.", "It is not a malware or document-macro scanner. Monument repeats every check — and malware, macro and archive-safety checks — independently after secure submission, and a Monument privacy reviewer inspects real excerpts before anyone else.", ] # ───────────────────────────────────────────────────────────── rules def load_rules(path=None): if path: with open(path, "r", encoding="utf-8") as fh: return json.load(fh) if not RULES_JSON.startswith("__MONUMENT"): return json.loads(RULES_JSON) here = os.path.dirname(os.path.abspath(__file__)) repo = os.path.join(here, "..", "..", "..", "config", "client-samples", "privacy-rules.json") with open(repo, "r", encoding="utf-8") as fh: return json.load(fh) def _flags(s): f = re.ASCII for ch in s or "": if ch == "i": f |= re.I elif ch == "m": f |= re.M elif ch == "s": f |= re.S return f class Compiled(object): def __init__(self, rules): self.rules = rules c = lambda d: (d, re.compile(d["pattern"], _flags(d.get("flags")))) self.secrets = [c(d) for d in rules["secrets"]] self.pii = [c(d) for d in rules["pii"]] self.regulated = [c(d) for d in rules["regulated"]] self.infra = [c(d) for d in rules["internalInfrastructure"]] self.names = [re.compile(p, re.ASCII) for p in rules["personName"]["patterns"]] self.name_value = re.compile(rules["personName"]["valuePattern"], re.ASCII) sk = rules["sensitiveKeys"] self.sensitive_keys = re.compile(sk["pattern"], _flags(sk.get("flags")) | re.M) self.placeholders = re.compile(sk["placeholders"], re.I | re.ASCII) self.special = re.compile(rules["specialCategoryTerms"], re.I | re.ASCII) self.git = re.compile(rules["gitHistory"], re.M | re.ASCII) self.conv = re.compile(rules["conversationTextMarkers"], re.I | re.M | re.ASCII) self.path = re.compile(rules["sensitivePathTerms"], re.I | re.ASCII) def luhn(digits): d = re.sub(r"\D", "", digits) if len(d) < 13 or len(d) > 19 or re.match(r"^(\d)\1+$", d): return False total = 0 for i, ch in enumerate(reversed(d)): n = int(ch) if i % 2 == 1: n *= 2 if n > 9: n -= 9 total += n return total % 10 == 0 def iban_valid(raw): s = re.sub(r"\s", "", raw).upper() if len(s) < 15 or len(s) > 34: return False moved = s[4:] + s[:4] rem = 0 for ch in moved: v = str(ord(ch) - 55) if "A" <= ch <= "Z" else ch for digit in v: if not digit.isdigit(): return False rem = (rem * 10 + int(digit)) % 97 return rem == 1 def count(rx, text, accept=None): n = 0 for m in rx.finditer(text): if accept is None or accept(m.group(0)): n += 1 return n def norm_key(k): return re.sub(r"[^a-z0-9]", "", k.lower()) def js_round(x): return int(x + 0.5) if x >= 0 else -int(-x + 0.5) def js_str(v): if isinstance(v, bool): return "true" if v else "false" if isinstance(v, float) and v.is_integer() and abs(v) < 1e21: return str(int(v)) return str(v) INVISIBLE = re.compile(u"[\u00ad\u034f\u061c\u115f\u1160\u17b4\u17b5\u180b-\u180f\u200b-\u200f\u202a-\u202e\u2060-\u206f\u3164\ufe00-\ufe0f\ufeff\uffa0]") def normalize_for_scan(text): """Fold compatibility forms (fullwidth digits), drop invisible format characters, unify line endings.""" return re.sub(r"\r\n?", "\n", INVISIBLE.sub("", unicodedata.normalize("NFKC", text))) def mask_manifest_identifiers(text): """Mask only the top-level package_id and the files' sha256 digests (when they have the expected shape).""" try: m = json.loads(text) except Exception: return text if not isinstance(m, dict): return text o = dict(m) if isinstance(o.get("package_id"), str) and re.match(r"^[A-Za-z0-9._-]{1,120}$", o["package_id"]): o["package_id"] = "[package id]" if isinstance(o.get("files"), list): o["files"] = [dict(f, sha256="[hash]") if isinstance(f, dict) and isinstance(f.get("sha256"), str) and re.match(r"^[0-9a-f]{64}$", f["sha256"]) else f for f in o["files"]] return json.dumps(o, indent=2, ensure_ascii=False) def mask_path_values(k, path): out = path for _, rx in k.secrets: out = rx.sub("[secret]", out) out = k.pii[0][1].sub("[email]", out) out = k.pii[1][1].sub("[phone]", out) return out def is_secret_like(path): return bool(SECRET_FILE.search(path)) def is_git_path(path): return ".git" in path.split("/") or bool(re.search(r"(^|/)\.git(modules|attributes|ignore)?$", path)) def structured_kind(path): p = path.lower() if p.endswith(".tsv"): return "tsv" if p.endswith(".csv"): return "csv" if p.endswith(".json"): return "json" if p.endswith(".jsonl") or p.endswith(".ndjson"): return "ndjson" return None def sample_area(path): if path == "sample_manifest.json": return "manifest" if path == "CHECKSUMS.sha256": return "checksums" if path.startswith(SAMPLE_DIRS["synthetic"]): return "synthetic" if path.startswith(SAMPLE_DIRS["authorized"]): return "authorized" if path.startswith(SAMPLE_DIRS["excerpts"]): return "excerpt" if path.startswith("SAMPLES/"): return "sample" if path in DOCUMENTS: return "documentation" return "other" # ───────────────────────────────────────────────────────────── privacy engine (port of privacy-core.ts) class Ctx(object): def __init__(self, k, mode): self.k = k self.rules = k.rules self.mode = mode self.linkage = {} self.findings = [] def add(self, code, severity, path, message): self.findings.append({"code": code, "severity": severity, "path": path, "message": message}) def link(self, key, rec): self.linkage.setdefault(key, set()).add(rec) def label(hits): return ", ".join("%s ×%d" % (l, n) for l, n in hits) def scan_text(text, path, area, ctx, from_decoded=False): r = ctx.k synthetic = area == "synthetic" secret_hits = [] for d, rx in r.secrets: n = count(rx, text) if n: secret_hits.append((d["label"], n)) config_hits = 0 for m in r.sensitive_keys.finditer(text): value = m.group(2) if r.sensitive_keys.groups >= 2 and m.group(2) is not None else "" value = re.sub(r"[;)\]]+$", "", value) if not r.placeholders.search(value) and not re.match(r"^\d{1,5}$", value): config_hits += 1 if secret_hits: ctx.add("encoded_secret" if from_decoded else "secret_detected", "sensitive", path, "%s (%s). Remove them and rotate them; never include secrets in a sample." % ("Encoded (base64/hex) content hides possible credentials" if from_decoded else "Possible credentials or keys", label(secret_hits))) if config_hits: ctx.add("sensitive_config", "sensitive", path, "%d setting(s) look like passwords, tokens, keys or connection strings with real values. Replace the values with placeholders, and rotate any that are real credentials." % config_hits) reg_hits = [] for d, rx in r.regulated: accept = luhn if d.get("luhn") else (iban_valid if d["id"] == "iban" else None) n = count(rx, text, accept) if n: reg_hits.append((d["label"], n)) if reg_hits: ctx.add("regulated_personal_data", "sensitive", path, "Possible regulated personal or financial identifiers (%s). These cannot be shared in a sample." % label(reg_hits)) if from_decoded: return 0, len(reg_hits) direct = [] for d, rx in r.pii: n = count(rx, text) if n and d.get("kind") == "direct": direct.append((d["label"], n)) names = sum(count(rx, text) for rx in r.names) if names: direct.append(("person name", names)) special = count(r.special, text) if direct: ctx.add("personal_name" if names and len(direct) == 1 else "personal_data_indicator", "review", path, "%s (%s). A privacy reviewer must confirm %s; remove or pseudonymise it if not." % ("Synthetic content that looks personal" if synthetic else "Possible personal details", label(direct), "it is fictional" if synthetic else "it may be shared")) if special and direct: ctx.add("sensitive_personal_data", "sensitive", path, "Health, financial or other special-category terms appear next to personal details. This kind of data cannot be shared in a sample.") elif special: ctx.add("special_category_terms", "review", path, "Health, financial or other special-category terms appear. A privacy reviewer must confirm no person can be identified.") infra = [] for d, rx in r.infra: n = count(rx, text) if n: infra.append((d["label"], n)) if infra: ctx.add("internal_infrastructure", "review", path, "Internal infrastructure details (%s). Remove hostnames, private addresses and account identifiers unless approved." % label(infra)) git = count(r.git, text) if git: ctx.add("git_history", "sensitive" if direct else "review", path, "Version-control history (%d marker(s): commits, authors, sign-offs or diffs)%s. Do not include Git history in a sample." % (git, " with personal details such as author emails" if direct else "")) convo = count(r.conv, text) if convo >= 2: ctx.add("conversation_content", "sensitive" if direct or reg_hits else "review", path, "Conversation-style content (%d turns)%s. Real conversations need explicit authorization and sanitization." % (convo, " containing personal details" if direct else "")) scan_encoded(text, path, area, ctx) return sum(n for _, n in direct), len(reg_hits) PRINTABLE = re.compile(r"[\x20-\x7e\s]+") def scan_encoded(text, path, area, ctx): lim = ctx.rules["limits"] b64 = re.compile(r"[A-Za-z0-9+/_-]{%d,}={0,2}" % lim["encodedMinLength"]) budget = lim["encodedMaxDecodeBytes"] for m in b64.finditer(text): if budget <= 0: break token = m.group(0) if re.match(r"^[0-9a-fA-F]+$", token) and len(token) % 2 == 0: decoded = bytes.fromhex(token).decode("latin-1") budget -= len(decoded) if re.fullmatch(r"[\x20-\x7e\s]{8,}", decoded): scan_text(decoded, path, area, ctx, from_decoded=True) continue t = token.replace("-", "+").replace("_", "/").rstrip("=") if len(t) % 4 == 1: t = t[:-1] t += "=" * ((4 - len(t) % 4) % 4) try: decoded = base64.b64decode(t).decode("utf-8", errors="replace") except Exception: continue budget -= len(decoded) if len(decoded) >= 8 and PRINTABLE.fullmatch(decoded): scan_text(decoded, path, area, ctx, from_decoded=True) def parse_csv(text, sep, max_rows): rows, row, field, quoted = [], [], "", False i, n = 0, len(text) while i < n and len(rows) < max_rows: ch = text[i] if quoted: if ch == '"' and i + 1 < n and text[i + 1] == '"': field += '"' i += 1 elif ch == '"': quoted = False else: field += ch elif ch == '"' and field == "": quoted = True elif ch == sep: row.append(field) field = "" elif ch == "\n" or ch == "\r": if ch == "\r" and i + 1 < n and text[i + 1] == "\n": i += 1 row.append(field) field = "" if any(c.strip() for c in row): rows.append(row) row = [] else: field += ch i += 1 if field or row: row.append(field) if any(c.strip() for c in row): rows.append(row) return rows def flatten(o, prefix="", out=None, depth=0): if out is None: out = {} if depth > 5: return out for k, v in o.items(): key = "%s.%s" % (prefix, k) if prefix else k if isinstance(v, dict): flatten(v, key, out, depth + 1) elif isinstance(v, (str, int, float, bool)) and v is not None: out[key] = js_str(v) return out def tables_from_json(value, out, depth=0): if depth > 6 or len(out) > 20: return if isinstance(value, list): objs_all = [v for v in value if isinstance(v, dict)] if objs_all: flat = [flatten(o) for o in objs_all[:2000]] cols = [] seen = set() for o in flat: for c in o.keys(): if c not in seen: seen.add(c) cols.append(c) out.append({"columns": cols[:300], "rows": flat}) for v in value[:200]: if isinstance(v, (dict, list)): tables_from_json(v, out, depth + 1) elif isinstance(value, dict): if depth == 0: f = flatten(value) out.append({"columns": list(f.keys()), "rows": [f]}) for v in value.values(): if isinstance(v, (dict, list)): tables_from_json(v, out, depth + 1) def has_conversation_shape(rules, value, depth=0): if depth > 6 or not isinstance(value, (dict, list)) or not value: return False if isinstance(value, list): return any(has_conversation_shape(rules, v, depth + 1) for v in value[:50]) keys = [norm_key(k) for k in value.keys()] if any(k in rules["conversationRoleKeys"] for k in keys) and any(k in ("content", "text", "message", "utterance", "body") for k in keys): return True if any(k in rules["conversationKeys"] for k in keys): return True return any(has_conversation_shape(rules, v, depth + 1) for v in list(value.values())[:50]) EMAIL_VALUE = re.compile(r"^[^@\s]+@[^@\s]+\.[A-Za-z]{2,}$") PERSON_COLUMN = re.compile(r"name|owner|by|contact|person|customer|employee|user", re.I) def profile_tables(tables, path, ctx): r = ctx.rules direct_h, quasi_h, id_h = set(r["headers"]["direct"]), set(r["headers"]["quasi"]), set(r["headers"]["personId"]) last_seg = lambda c: norm_key(c.split(".")[-1]) for ti, t in enumerate(tables): if not t["rows"]: continue cols = t["columns"] direct = [c for c in cols if last_seg(c) in direct_h or norm_key(c) in direct_h] quasi = [c for c in cols if last_seg(c) in quasi_h] ids = [c for c in cols if last_seg(c) in id_h] value_direct = [] for c in cols: if c in direct: continue vals = [row.get(c, "") for row in t["rows"]] vals = [v for v in vals if v.strip()][:200] if len(vals) < 2: continue email = sum(1 for v in vals if EMAIL_VALUE.match(v.strip())) named = sum(1 for v in vals if ctx.k.name_value.search(v.strip())) if email / float(len(vals)) >= 0.5 or (named / float(len(vals)) >= 0.8 and PERSON_COLUMN.search(c)): value_direct.append(c) identifying = direct + value_direct for ri, row in enumerate(t["rows"]): rec = "%s#%d:%d" % (path, ti, ri) for c in ids + identifying: v = row.get(c, "").strip().lower() if len(v) >= 3: ctx.link("%s:%s:%s" % ("id" if c in ids else "direct", last_seg(c), v), rec) rows = len(t["rows"]) limit = r["limits"]["maxRecordsPerTable"].get(ctx.mode) if ctx.mode else None if identifying and rows >= r["limits"]["recordsThreshold"]: ctx.add("personal_records", "sensitive", path, "%d records with identifying fields (%d column(s) such as names, emails, phones or addresses). Customer, employee or user records cannot be shared as a sample; describe them in the data dictionary instead, or use synthetic rows." % (rows, len(identifying))) elif identifying: ctx.add("personal_data_indicator", "review", path, "A record with identifying fields (%d column(s)). A privacy reviewer must confirm it may be shared." % len(identifying)) if len(quasi) >= r["limits"]["quasiIdentifierThreshold"]: ctx.add("quasi_identifiers", "review", path, "Records combine %d indirect identifiers (for example postcode, birth date, gender, job title). Together these can identify a person; generalise or remove some of them." % len(quasi)) if limit is not None and rows > limit: ctx.add("too_much_sample_content", "correction", path, "%d records in one table; samples in this mode may contain at most %d. Include a small, representative subset." % (rows, limit)) def key_line_identifiers(line, path, index, ctx): direct = 0 for d, rx in ctx.k.pii: if d.get("kind") != "direct": continue for m in rx.finditer(line): direct += 1 ctx.link("%s:%s" % (d["id"], re.sub(r"[\s().-]", "", m.group(0).lower())), "%s#line:%d" % (path, index)) return direct def status_of(findings): sev = set(f["severity"] for f in findings) if "sensitive" in sev: return "SENSITIVE_DATA_FOUND" if "unsupported" in sev: return "UNSUPPORTED_CONTENT" if "incomplete" in sev: return "SCAN_INCOMPLETE" if "review" in sev: return "REVIEW_REQUIRED" return "CHECKS_PASSED" def dedupe(findings): seen, out = set(), [] for f in findings: key = (f["code"], f["severity"], f["path"], f["message"]) if key in seen: continue seen.add(key) out.append(f) return out def scan_name_values(path, raw, ctx): k = ctx.k if sum(count(rx, path) for _, rx in k.secrets): ctx.add("secret_in_name", "sensitive", raw, "A file or folder name looks like a credential or key. Rename it, and rotate the credential if it is real.") reg = 0 for d, rx in k.regulated: reg += count(rx, path, luhn if d.get("luhn") else (iban_valid if d["id"] == "iban" else None)) if reg: ctx.add("regulated_in_name", "sensitive", raw, "A file or folder name contains what looks like a regulated personal or financial identifier. Rename it.") def assess_privacy(k, files, mode, archive_comment, receiving=False, extra_paths=None, budget_total=None, budget_entry=None): """files: dicts {path, bytes, area, kind: text|unsupported|unreadable|missing, text?, typeLabel?, metadata?, structured?}""" r = k.rules ctx = Ctx(k, mode) budget = budget_total or SCAN_TOTAL per_entry = budget_entry or SCAN_PER_ENTRY cov = {"scannedFiles": 0, "scannedBytes": 0, "unsupported": 0, "incomplete": 0, "rulesVersion": r["rulesVersion"]} def incomplete(path, why): cov["incomplete"] += 1 ctx.add("scan_incomplete", "incomplete", path, "%s This file was not fully checked, so the sample cannot be cleared automatically. Reduce or split it." % why) if archive_comment and archive_comment.strip(): ctx.add("archive_metadata", "review", None, "The ZIP carries a comment. Comments can reveal who made it or where it came from; create the ZIP without one.") scan_text(normalize_for_scan(archive_comment), "(ZIP comment)", "other", ctx) samples = [f for f in files if f["area"] in ("synthetic", "authorized", "excerpt", "sample")] if mode and mode != "metadata_only": max_files = r["limits"]["maxSampleFiles"].get(mode, 0) max_bytes = r["limits"]["maxSampleBytes"].get(mode, 0) total = sum(f["bytes"] for f in samples) if len(samples) > max_files: ctx.add("too_much_sample_content", "correction", "SAMPLES/", "%d sample files; this mode allows at most %d. Include a few short, representative examples — never a whole repository or dataset." % (len(samples), max_files)) if total > max_bytes: ctx.add("too_much_sample_content", "correction", "SAMPLES/", "Sample files total %d KB; this mode allows at most %d KB." % (js_round(total / 1024.0), js_round(max_bytes / 1024.0))) real = len([f for f in samples if f["area"] in ("excerpt", "authorized")]) if receiving and real: ctx.add("human_privacy_review_required", "review", "SAMPLES/", "The sample contains %d real excerpt or authorized example file(s). A Monument privacy reviewer inspects real content before anyone else; automated checks cannot establish that it is free of personal or confidential data." % real) source = len([f for f in samples if SOURCE_EXT.search(f["path"])]) markers = len([f for f in samples if re.search(r"(^|/)(package\.json|pyproject\.toml|setup\.py|go\.mod|Cargo\.toml|pom\.xml|build\.gradle|Gemfile|composer\.json|requirements\.txt|Makefile|Dockerfile)$", f["path"], re.I)]) if source >= 40 or (markers and source >= 15): ctx.add("bulk_content", "review", "SAMPLES/", "The sample looks like a whole source tree (%d source files%s). Samples must be short, selected excerpts." % (source, ", project build files" if markers else "")) def check_path(raw, is_file): np_ = normalize_for_scan(raw) if k.path.search(np_): ctx.add("sensitive_path", "review", raw, "The file or folder name suggests confidential or personal data (for example payroll, customers, contracts). Rename it, or remove the file if it is what the name says.") if count(k.pii[0][1], np_) + count(k.pii[1][1], np_): ctx.add("sensitive_path", "sensitive", raw, "A file or folder name contains an email address or phone number. Rename it.") if not is_file: scan_name_values(np_, raw, ctx) for xp in extra_paths or []: check_path(xp, False) for f in files: p = f["path"] check_path(p, True) kind = f["kind"] if kind == "missing": incomplete(p, "The file could not be read (encrypted, damaged, or too large to extract).") continue if kind == "unsupported": cov["unsupported"] += 1 ctx.add("unsupported_content", "unsupported", p, "This file type (%s) cannot be checked for personal or confidential data automatically (images, PDFs, audio, video and binary files can contain faces, names, signatures or location data). Replace it with a text description, or a Monument privacy reviewer must inspect it." % f.get("typeLabel", "unknown")) continue if kind == "unreadable" or f.get("text") is None: cov["unsupported"] += 1 ctx.add("unreadable_content", "unsupported", p, "This file is not valid text, so it cannot be checked. Save it as UTF-8 text or remove it.") continue metadata = f.get("metadata") or [] if metadata: ctx.add("document_metadata", "review", p, "The document's properties carry %d value(s) such as author or company names. Remove document properties before sharing." % len(metadata)) for m in metadata: scan_text(normalize_for_scan(m), p, f["area"], ctx) text = normalize_for_scan(mask_manifest_identifiers(f["text"]) if f["area"] == "manifest" else f["text"]) allowed = max(0, min(len(text), per_entry, budget)) if allowed < len(text): incomplete(p, "Only %d KB of %d KB could be scanned within the checking budget." % (js_round(allowed / 1024.0), js_round(len(text) / 1024.0))) body = text[:allowed] budget -= allowed cov["scannedBytes"] += allowed cov["scannedFiles"] += 1 tables, conversation = [], False sk = f.get("structured") if sk in ("csv", "tsv"): rows = parse_csv(body, "\t" if sk == "tsv" else ",", 5001) if len(rows) >= 2: header = [c.strip() for c in rows[0]] recs = [] for row in rows[1:]: rec = {} for i, c in enumerate(header): rec[c] = row[i] if i < len(row) else "" recs.append(rec) tables.append({"columns": header, "rows": recs}) elif sk == "json": try: v = json.loads(body) if f["area"] != "manifest": tables_from_json(v, tables) conversation = has_conversation_shape(r, v) except Exception: pass elif sk == "ndjson": objs = [] for line in re.split(r"\r?\n", body)[:5000]: if not line.strip(): continue try: objs.append(json.loads(line)) except Exception: pass tables_from_json(objs, tables) conversation = has_conversation_shape(r, objs) if tables: profile_tables(tables, p, ctx) lines = re.split(r"\r?\n", body) direct = 0 for i, line in enumerate(lines): if line.strip(): direct += key_line_identifiers(line, p, i, ctx) d2, regulated = scan_text(body, p, f["area"], ctx) direct = max(direct, d2) if conversation: ctx.add("conversation_content", "sensitive" if direct or regulated else "review", p, "Conversation or agent-trajectory records%s. Real conversations need explicit authorization and sanitization; prefer synthetic ones." % (" containing personal details" if direct or regulated else "")) min_occ = r["limits"]["linkageMinOccurrences"] linked_direct = linked_ids = 0 for key, recs in ctx.linkage.items(): if len(recs) < min_occ: continue files_of = set(x.split("#")[0] for x in recs) if key.startswith("id:") and len(files_of) < 2: continue if key.startswith("id:"): linked_ids += 1 else: linked_direct += 1 if linked_direct: ctx.add("linked_identifiers", "sensitive", None, "%d personal identifier(s) (such as an email address or phone number) appear in several records or files, which links them to one person." % linked_direct) if linked_ids: ctx.add("linked_identifiers", "review", None, "%d person identifier(s) (such as a customer id) appear in more than one file, so the files can be joined into a profile. Replace them with unrelated pseudonyms per file, or remove the join." % linked_ids) out = dedupe(ctx.findings) return status_of(out), out, cov # ───────────────────────────────────────────────────────────── reading files def sniff(content): head = content[:16] if head.startswith(b"%PDF"): return "pdf" if head.startswith(b"\x89PNG") or head[:3] == b"\xff\xd8\xff" or head[:6] in (b"GIF87a", b"GIF89a") or head[:4] == b"II*\x00" or head[:4] == b"MM\x00*" or (head[:4] == b"RIFF" and content[8:12] == b"WEBP"): return "image" if head[:4] == b"RIFF" or head[:4] == b"OggS" or head[:3] == b"ID3" or head[:4] == b"fLaC": return "audio" if content[4:8] == b"ftyp" or head[:4] == b"\x1aE\xdf\xa3": return "video" if head[:2] == b"MZ" or head[:4] == b"\x7fELF" or head[:4] in (b"\xfe\xed\xfa\xce", b"\xfe\xed\xfa\xcf", b"\xce\xfa\xed\xfe", b"\xcf\xfa\xed\xfe", b"\xca\xfe\xba\xbe"): return "executable" if head[:4] == b"PK\x03\x04": return "zip" if head[:2] == b"\x1f\x8b" or head[:6] == b"7z\xbc\xaf\x27\x1c" or head[:4] == b"Rar!" or content[257:262] == b"ustar": return "archive" if b"\x00" in content[:8192]: return "binary" try: content[:65536].decode("utf-8") except UnicodeDecodeError as e: if e.start < 65536 - 4: return "binary" return "text" def decode_strict(content): try: return content.decode("utf-8-sig") except UnicodeDecodeError: return None def ooxml_text(content): import io try: zf = zipfile.ZipFile(io.BytesIO(content)) infos = zf.infolist() if len(infos) > 2000: return None text, metadata, total = [], [], 0 for info in infos: name = info.filename if not (OOXML_TEXT_PART.match(name) or OOXML_META_PART.match(name)): continue if info.flag_bits & 0x1 or info.file_size > MAX_ENTRY_BYTES or (info.file_size > RATIO_FLOOR and info.compress_size and info.file_size / float(info.compress_size) > MAX_RATIO): return None total += info.file_size if total > MAX_UNCOMPRESSED: return None xml = zf.read(info).decode("utf-8", errors="replace") if OOXML_META_PART.match(name): for m in re.finditer(r"<(?:dc:creator|cp:lastModifiedBy|Company|Manager|dc:title|cp:keywords|dc:description)[^>]*>([^<]{1,300})<", xml): if m.group(1).strip(): metadata.append(m.group(1).strip()) continue t = re.sub(r"", "\n", xml) t = re.sub(r"<[^>]+>", " ", t).replace("<", "<").replace(">", ">").replace("&", "&").replace(""", '"') text.append(re.sub(r"[ \t]+", " ", t)) return "\n".join(text), metadata except Exception: return None BLANK_FORK = b"This resource fork intentionally left blank" BENIGN_XATTRS = {"com.apple.quarantine": 256, "com.apple.provenance": 64, "com.apple.lastuseddate#PS": 32, "com.apple.TextEncoding": 64, "com.apple.metadata:kMDItemDownloadedDate": 128} def apple_double_is_benign(b): """An AppleDouble file exactly as macOS writes it for a ZIP (port of appleDoubleIsBenign in privacy-core.ts).""" n_b = len(b) u32 = lambda o: struct.unpack(">I", b[o:o + 4])[0] if o + 4 <= n_b else -1 u16 = lambda o: struct.unpack(">H", b[o:o + 2])[0] if o + 2 <= n_b else -1 if n_b < 26 or n_b > 65536 or u32(0) != 0x00051607 or u32(4) != 0x00020000: return False covered = bytearray(n_b) def cover(o, n): for i in range(max(0, o), min(n_b, o + n)): covered[i] = 1 for i in range(8, 24): if b[i] != 0 and (b[i] < 0x20 or b[i] > 0x7E): return False n = u16(24) if n < 0 or n > 4: return False cover(0, 26 + n * 12) for i in range(n): at = 26 + i * 12 eid, off, ln = u32(at), u32(at + 4), u32(at + 8) if off < 0 or ln < 0 or off + ln > n_b: return False if eid == 2: if ln == 0: continue chunk = b[off:off + ln] if ln != 286 or BLANK_FORK not in chunk: return False rest = chunk.replace(BLANK_FORK, b"", 1) if any(0x20 <= c <= 0x7E for c in rest): return False cover(off, ln) elif eid == 9: if ln < 32: return False cover(off, 32) if ln == 32: continue a = off + 34 if u32(a) != 0x41545452: return False cnt = u16(a + 34) if cnt < 0 or cnt > 8: return False cover(off + 32, 2 + 36) p = a + 36 for _ in range(cnt): v_off, v_len = u32(p), u32(p + 4) if p + 10 >= n_b: return False name_len = b[p + 10] if v_off < 0 or v_len < 0 or p + 11 + name_len > n_b: return False name = b[p + 11:p + 11 + name_len].decode("latin-1").rstrip("\x00") mx = BENIGN_XATTRS.get(name) if mx is None or v_len > mx or v_off + v_len > n_b: return False value = b[v_off:v_off + v_len].rstrip(b"\x00") if name in ("com.apple.quarantine", "com.apple.TextEncoding") and any(c < 0x20 or c > 0x7E for c in value): return False entry_len = (11 + name_len + 3) & ~3 cover(p, entry_len) cover(v_off, v_len) p += entry_len else: return False for i in range(n_b): if not covered[i] and b[i] != 0: return False return True def os_metadata_ok(path, content): """Only standard AppleDouble files (macOS ZIP metadata) are listed and ignored; .DS_Store never is.""" if not OS_METADATA_NAME.search(path) or content is None: return False base = path.split("/")[-1] if not base.startswith("._"): return False return len(content) == 0 or apple_double_is_benign(content) def printable_strings(content): return "\n".join(m.group(0).decode("ascii") for m in re.finditer(rb"[\x20-\x7e]{8,}", content)) def to_core(path, content, area): """What the rules engine can read from one file.""" f = {"path": path, "bytes": len(content) if content is not None else 0, "area": area} if content is None: f["kind"] = "missing" return f t = sniff(content) f["typeLabel"] = t if OOXML_EXT.search(path): x = ooxml_text(content) if x is None: f.update(kind="unsupported", typeLabel="Office document that could not be opened") else: f.update(kind="text", text=x[0], metadata=x[1], structured=None) return f is_text = t == "text" or TEXT_EXT.search(path) or SOURCE_EXT.search(path) if not is_text: f["kind"] = "unsupported" return f text = decode_strict(content) if text is None: f["kind"] = "unreadable" return f f.update(kind="text", text=text, structured=structured_kind(path)) return f # ───────────────────────────────────────────────────────────── package contract (port of contract.ts) ENUMS = { "asset_type": ["codebase", "structured_data", "documents_text", "agent_trajectories", "rl_environment", "ai_training_data", "multimodal", "other"], "sample_mode": ["metadata_only", "synthetic_or_authorized_examples", "real_excerpts"], } def manifest_problems(m): """A compact structural validation of sample_manifest.json 3.0 (field names only, never values).""" bad = [] def need(obj, key, kind, where, allowed=None, nullable=False): if not isinstance(obj, dict) or key not in obj: bad.append(where) return None v = obj[key] if v is None: if not nullable: bad.append(where) return None ok = {"str": isinstance(v, str), "bool": isinstance(v, bool), "int": isinstance(v, int) and not isinstance(v, bool), "obj": isinstance(v, dict), "list": isinstance(v, list)}[kind] if not ok or (allowed and v not in allowed): bad.append(where) return None return v allowed_top = {"schema_version", "prompt_version", "package_id", "asset_type", "sample_mode", "generated_at", "generated_with", "processing", "intended_recipient", "source_summary", "rights", "privacy", "disclosure_authorization", "authorized_examples", "local_checker", "rl_environment", "local_checks", "owner_approval", "files", "limitations"} nested = {"processing": {"environment", "assistant_cloud_connected", "organization_approval"}, "source_summary": {"description", "population_count", "population_count_status", "sampling_method"}, "rights": {"ownership", "third_party_content", "ai_training_permission", "verified_by_monument"}, "privacy": {"personal_data_in_sample", "human_review_complete"}, "disclosure_authorization": {"approved_by_role", "approved_at", "basis", "scope", "excerpt_count", "sanitization_applied", "re_checked_after_sanitization"}, "local_checker": {"tool", "version", "candidate_status"}, "rl_environment": {"runnable_claimed", "execution_evidence_included"}, "owner_approval": {"status", "approved_by_role", "approved_at", "scope"}} if isinstance(m, dict): for key in m: if key not in allowed_top: bad.append(key[:40]) for key, keys in nested.items(): if isinstance(m.get(key), dict) and set(m[key]) - keys: bad.append(key) for key, keys in (("files", {"path", "sha256", "bytes", "role"}), ("authorized_examples", {"path", "authorization_basis"}), ("local_checks", {"name", "result", "limits"})): if isinstance(m.get(key), list) and any(isinstance(x, dict) and set(x) - keys for x in m[key]): bad.append(key) need(m, "package_id", "str", "package_id") need(m, "asset_type", "str", "asset_type", ENUMS["asset_type"]) need(m, "sample_mode", "str", "sample_mode", ENUMS["sample_mode"]) need(m, "intended_recipient", "str", "intended_recipient") pr = need(m, "processing", "obj", "processing") if pr is not None: need(pr, "environment", "str", "processing.environment", ["offline_local", "approved_ai_assistant", "manual"]) if "assistant_cloud_connected" not in pr or not (pr["assistant_cloud_connected"] is None or isinstance(pr["assistant_cloud_connected"], bool)): bad.append("processing.assistant_cloud_connected") need(pr, "organization_approval", "str", "processing.organization_approval", ["confirmed", "not_required", "unknown"]) ss = need(m, "source_summary", "obj", "source_summary") if ss is not None: need(ss, "description", "str", "source_summary.description") need(ss, "population_count_status", "str", "source_summary.population_count_status", ["measured", "estimated", "owner_asserted", "unknown"]) need(ss, "sampling_method", "str", "source_summary.sampling_method") rt = need(m, "rights", "obj", "rights") if rt is not None: need(rt, "ownership", "str", "rights.ownership", ["owner_asserted", "unknown", "disputed"]) need(rt, "third_party_content", "str", "rights.third_party_content", ["none_known", "present", "unknown"]) need(rt, "ai_training_permission", "str", "rights.ai_training_permission", ["owner_asserted", "not_permitted", "unknown"]) need(rt, "verified_by_monument", "bool", "rights.verified_by_monument") pv = need(m, "privacy", "obj", "privacy") if pv is not None: need(pv, "personal_data_in_sample", "str", "privacy.personal_data_in_sample", ["none_known", "present_reviewed", "unknown"]) need(pv, "human_review_complete", "bool", "privacy.human_review_complete") if "disclosure_authorization" not in m: bad.append("disclosure_authorization") elif m["disclosure_authorization"] is not None: da = m["disclosure_authorization"] for key, kind in (("approved_by_role", "str"), ("approved_at", "str"), ("basis", "str"), ("scope", "str"), ("excerpt_count", "int"), ("sanitization_applied", "list"), ("re_checked_after_sanitization", "bool")): need(da, key, kind, "disclosure_authorization." + key) if need(m, "authorized_examples", "list", "authorized_examples") is not None: for x in m["authorized_examples"]: if not isinstance(x, dict) or not isinstance(x.get("path"), str) or not isinstance(x.get("authorization_basis"), str) or len(x.get("authorization_basis", "")) < 10: bad.append("authorized_examples") break if "local_checker" not in m: bad.append("local_checker") elif m["local_checker"] is not None: need(m["local_checker"], "candidate_status", "str", "local_checker.candidate_status", ["CHECKS_PASSED", "SENSITIVE_DATA_FOUND", "REVIEW_REQUIRED", "SCAN_INCOMPLETE", "UNSUPPORTED_CONTENT"]) need(m, "local_checks", "list", "local_checks") oa = need(m, "owner_approval", "obj", "owner_approval") if oa is not None: need(oa, "status", "str", "owner_approval.status", ["approved", "pending", "declined"]) if need(m, "files", "list", "files") is not None: for x in m["files"]: if not isinstance(x, dict) or not isinstance(x.get("path"), str) or not isinstance(x.get("sha256"), str) or not SHA.match(x.get("sha256", "")) or not isinstance(x.get("bytes"), int) or x.get("role") not in ("documentation", "synthetic_example", "authorized_example", "real_excerpt"): bad.append("files") break need(m, "limitations", "list", "limitations") return sorted(set(bad))[:8] ROLE_OF_AREA = {"documentation": "documentation", "synthetic": "synthetic_example", "authorized": "authorized_example", "excerpt": "real_excerpt"} def parse_checksums(text): entries, invalid, duplicates = {}, 0, 0 for line in re.split(r"\r?\n", text): if not line.strip(): continue m = re.match(r"^([0-9a-fA-F]{64}) [ *](.+)$", line) if not m: invalid += 1 continue path = re.sub(r"^\./", "", m.group(2).strip()).replace("\\", "/") if path in entries: duplicates += 1 entries[path] = m.group(1).lower() return entries, invalid, duplicates def validate_contract(files): """files: list of {path, bytes, sha256, content}. Returns (findings, manifest dict or None).""" out = [] add = lambda code, sev, path, msg: out.append({"code": code, "severity": sev, "path": path, "message": msg}) by = dict((f["path"], f) for f in files) legacy = [f for f in LEGACY_V1_FILES if f in by] for req in REQUIRED_FILES: if req not in by: add("missing_required_file", "correction", req, "%s is missing. The current Monument prompt creates it; regenerate the package with that prompt." % req) for f in files: area = sample_area(f["path"]) if area == "other" and f["path"] not in legacy: add("unexpected_file", "correction", f["path"], "This file is not part of the sample package. Only the documented files and the SAMPLES/ folders belong in the ZIP.") if area == "sample": add("unexpected_sample_folder", "correction", f["path"], "Sample files must be in SAMPLES/SYNTHETIC/ (fictional), SAMPLES/AUTHORIZED/ (authorized examples) or SAMPLES/EXCERPTS/ (approved real excerpts).") for doc in DOCUMENTS: f = by.get(doc) if not f: continue if f["bytes"] > MAX_DOCUMENT_BYTES: add("document_too_large", "correction", doc, "%s is unusually large. Keep documents short and descriptive." % doc) elif f["content"] is None or len(f["content"].decode("utf-8", errors="replace").strip()) < 20: add("empty_document", "correction", doc, "%s is empty or almost empty." % doc) manifest = None mf = by.get("sample_manifest.json") if mf: if mf["bytes"] > MAX_MANIFEST_BYTES or mf["content"] is None: add("manifest_too_large", "correction", mf["path"], "sample_manifest.json is too large to be a manifest.") else: raw = None try: raw = json.loads(mf["content"].decode("utf-8")) except Exception: add("manifest_invalid_json", "correction", mf["path"], "sample_manifest.json is not valid JSON.") if raw is not None: version = raw.get("schema_version") if isinstance(raw, dict) else None if version in ("1.0", "2.0") or legacy: add("legacy_contract", "correction", mf["path"], "This package uses an older sample format (%s). Regenerate it with the current Monument prompt (format %s), which adds the sample modes and the privacy checks." % (version if isinstance(version, str) else "1.0", CONTRACT_VERSION)) elif version != CONTRACT_VERSION: add("unsupported_manifest_version", "correction", mf["path"], 'sample_manifest.json must declare "schema_version": "%s".' % CONTRACT_VERSION) else: problems = manifest_problems(raw) if problems: add("manifest_invalid", "correction", mf["path"], "sample_manifest.json does not match the required structure (check: %s)." % ", ".join(problems)) else: manifest = raw if manifest is not None: m = manifest if m["intended_recipient"].strip() != "Monument Exchange": add("wrong_recipient", "correction", "sample_manifest.json", 'The manifest must name "Monument Exchange" as the intended recipient.') if m["owner_approval"]["status"] != "approved": add("owner_approval_missing", "correction", "sample_manifest.json", "The manifest says the owner has not approved this package for Monument Exchange. Review it with your assistant and approve it before zipping.") if m["rights"]["verified_by_monument"]: add("false_verification_claim", "correction", "sample_manifest.json", 'A package cannot declare Monument verification; set "verified_by_monument" to false.') if m["processing"]["environment"] == "approved_ai_assistant" and m["processing"]["organization_approval"] != "confirmed": add("assistant_not_approved", "correction", "sample_manifest.json", "The package was prepared with an AI assistant whose approval by your organization is not confirmed. Use only an approved assistant, or prepare the package offline.") lc = m.get("local_checker") if lc and lc.get("candidate_status") in ("SENSITIVE_DATA_FOUND", "SCAN_INCOMPLETE"): add("local_check_not_passed", "correction", "sample_manifest.json", "The offline checker found possible sensitive data or could not scan the candidate folder. Fix its findings and re-run it before zipping.") elif lc and lc.get("candidate_status") in ("UNSUPPORTED_CONTENT", "REVIEW_REQUIRED"): add("local_check_needs_review", "warning", "sample_manifest.json", "The offline checker asked for a human review of the candidate folder. Monument's privacy reviewer will review it before anyone else.") if not lc: add("local_checker_not_run", "warning", "sample_manifest.json", "The package does not record a run of the offline Monument sample checker. Monument checks it independently anyway.") listed = {} for item in m["files"]: if item["path"] in listed: add("manifest_duplicate_file", "correction", item["path"], "The manifest lists this file twice.") listed[item["path"]] = item f = by.get(item["path"]) if not f: add("manifest_file_missing", "correction", item["path"], "The manifest lists a file that is not in the ZIP.") elif f["sha256"] and (f["sha256"] != item["sha256"] or f["bytes"] != item["bytes"]): add("manifest_hash_mismatch", "correction", item["path"], "This file's size or SHA-256 does not match the manifest. Recompute the manifest after the last edit.") expected = ROLE_OF_AREA.get(sample_area(item["path"])) if expected and item["role"] != expected: add("role_mismatch", "correction", item["path"], 'Files in this folder must have the role "%s".' % expected) for f in files: if f["path"] in ("sample_manifest.json", "CHECKSUMS.sha256"): continue if f["path"] not in listed: add("file_not_in_manifest", "correction", f["path"], "This file is in the ZIP but not listed in sample_manifest.json.") samples = [f for f in files if f["path"].startswith("SAMPLES/")] of = lambda a: [f for f in samples if sample_area(f["path"]) == a] synthetic, authorized, excerpts = of("synthetic"), of("authorized"), of("excerpt") mode = m["sample_mode"] if mode == "metadata_only" and samples: add("metadata_only_has_samples", "correction", "SAMPLES/", "The package is metadata only but contains a SAMPLES/ folder. Remove it, or choose another sample mode with the required authorization.") if mode == "synthetic_or_authorized_examples" and excerpts: add("excerpts_need_real_mode", "correction", SAMPLE_DIRS["excerpts"], 'Real excerpts (%s) need sample mode "real_excerpts" and a disclosure authorization.' % SAMPLE_DIRS["excerpts"]) if mode == "synthetic_or_authorized_examples" and not synthetic and not authorized: add("no_samples_included", "warning", None, "The package says it includes examples but SAMPLES/ is empty.") if mode == "real_excerpts": a = m["disclosure_authorization"] if not a: add("disclosure_authorization_missing", "correction", "sample_manifest.json", "Real excerpts need an explicit disclosure authorization (who approved, on what basis, what scope, which sanitization).") else: if not a["re_checked_after_sanitization"]: add("not_rechecked_after_sanitization", "correction", "sample_manifest.json", "Re-run the checks after sanitizing the excerpts and record it.") if a["excerpt_count"] != len(excerpts): add("excerpt_count_mismatch", "correction", "sample_manifest.json", "The authorization covers %d excerpt(s) but the package contains %d." % (a["excerpt_count"], len(excerpts))) if not m["privacy"]["human_review_complete"]: add("privacy_review_missing", "correction", "sample_manifest.json", "Real excerpts need a completed human privacy review before zipping.") if not excerpts: add("no_samples_included", "warning", None, "The package says it includes real excerpts but SAMPLES/EXCERPTS/ is empty.") bases = set(x["path"] for x in m["authorized_examples"]) for f in authorized: if f["path"] not in bases: add("authorization_basis_missing", "correction", f["path"], "An authorized example needs its authorization basis in the manifest (authorized_examples).") # Monument stays blind to seller source code (ADR-0024): codebases are described, never sampled. if m["asset_type"] == "codebase" and mode != "metadata_only": add("codebase_metadata_only", "correction", "sample_manifest.json", 'Codebases are described, never sampled: Monument does not receive source code. Use sample mode "metadata_only" here, and Monument Scrub for the code itself.') for f in excerpts + authorized: if SOURCE_EXT.search(f["path"]): add("source_code_not_accepted", "correction", f["path"], "Monument does not receive seller source code. Remove this file and describe the code in the documents instead (Monument Scrub covers code).") if synthetic and not any(re.search("synthetic", l, re.I) for l in m["limitations"] if isinstance(l, str)): add("synthetic_disclaimer_missing", "correction", "sample_manifest.json", "Synthetic examples show format only. Add a limitation saying they are synthetic and say nothing about the quality or characteristics of the real data.") cf = by.get("CHECKSUMS.sha256") if cf and cf["content"] is not None: entries, invalid, duplicates = parse_checksums(cf["content"].decode("utf-8", errors="replace")) if invalid: add("checksums_invalid", "correction", cf["path"], 'CHECKSUMS.sha256 has %d line(s) that are not " ".' % invalid) if duplicates: add("checksums_duplicate_path", "correction", cf["path"], "CHECKSUMS.sha256 lists %d path(s) more than once. List every file exactly once." % duplicates) for path, sha in entries.items(): f = by.get(path) if not f: add("checksum_file_missing", "correction", path[:300], "CHECKSUMS.sha256 lists a file that is not in the ZIP.") elif f["sha256"] and f["sha256"] != sha: add("checksum_mismatch", "correction", path, "This file's SHA-256 does not match CHECKSUMS.sha256. Recompute the checksums after the last edit.") for f in files: if f["path"] != "CHECKSUMS.sha256" and f["path"] not in entries: add("file_not_in_checksums", "correction", f["path"], "This file is not listed in CHECKSUMS.sha256.") return out, manifest # ───────────────────────────────────────────────────────────── gathering a package (folder or ZIP) def classify_entry(rel, content, add): """Name-level checks shared by folders and ZIPs. Returns False when the file must not be read as package content.""" segments = rel.split("/") if is_secret_like(rel): add("secret_file", "sensitive", rel, "This looks like a credentials, key or shell-history file. Never include secrets in a sample.") elif DEPENDENCY_DIR.search(rel): add("excluded_folder", "correction", rel, "Version-control, dependency or build folders do not belong in a sample. Remove them.") elif any(s.startswith(".") for s in segments): add("hidden_file", "correction", rel, "Hidden files do not belong in a sample. Remove them.") if is_git_path(rel): add("git_history", "sensitive", rel, "Version-control data (.git) carries the full history, author names and emails. Never include it in a sample.") if EXECUTABLE_EXT.search(rel): add("executable_file", "security", rel, "The package contains a program or disk-image file. Samples must not contain executables.") elif ARCHIVE_EXT.search(rel): add("nested_archive", "correction", rel, "ZIP or other archive files inside the sample are not accepted. Include the files themselves.") if content is not None: t = sniff(content) if t == "executable": add("executable_binary", "security", rel, "A file contains executable program code. Samples must not contain programs.") elif t in ("zip", "archive") and not ARCHIVE_EXT.search(rel) and not OOXML_EXT.search(rel): add("nested_archive", "correction", rel, "A file in the sample is itself an archive. Include the files themselves.") if re.search(r"[\x00-\x1f\x7f]", rel) or re.search(u"[‪-‮⁦-⁩]", rel): add("control_characters_in_name", "security", rel, "A file name contains control or direction-changing characters.") return True def gather_folder(root, add): """Walk a staged folder without following links. Returns [(rel, content|None)].""" items = [] if not os.path.isdir(root): raise SystemExit("not a folder: %s" % root) count_files = 0 for dirpath, dirnames, filenames in os.walk(root, followlinks=False): for d in list(dirnames): full = os.path.join(dirpath, d) if os.path.islink(full): add("symlink", "security", os.path.relpath(full, root).replace(os.sep, "/"), "A symbolic link is not allowed in a sample.") dirnames.remove(d) dirnames.sort() for name in sorted(filenames): full = os.path.join(dirpath, name) rel = os.path.relpath(full, root).replace(os.sep, "/") st = os.lstat(full) if stat.S_ISLNK(st.st_mode): add("symlink", "security", rel, "A symbolic link is not allowed in a sample.") continue if not stat.S_ISREG(st.st_mode): add("special_file", "security", rel, "A device, pipe or other special file is not allowed in a sample.") continue count_files += 1 if count_files > MAX_ENTRIES: add("too_many_entries", "correction", None, "More than %d files. Include fewer, representative files." % MAX_ENTRIES) add("scan_incomplete", "incomplete", None, "Not every file was checked because the folder holds too many files.") return items if st.st_size > MAX_ENTRY_BYTES: add("too_large", "correction", rel, "This file is larger than samples allow.") items.append((rel, None)) continue with open(full, "rb") as fh: items.append((rel, fh.read())) return items def _extra_unicode_path(extra): i = 0 while i + 4 <= len(extra): hid, size = struct.unpack("= 5: return body[5:].decode("utf-8", errors="replace") i += 4 + size return None KNOWN_EXTRA_IDS = {0x0001, 0x000A, 0x5455, 0x7875, 0x7855, 0x7075, 0x6375, 0x5855, 0xA220, 0xD935} def gather_zip(path, add, max_entry=MAX_ENTRY_BYTES, max_total=MAX_UNCOMPRESSED): """Inspect every entry of the final ZIP. Returns ([(rel_with_root, content|None)], comment).""" with open(path, "rb") as fh: raw = fh.read() if not raw.startswith(b"PK\x03\x04"): if raw[:4] == b"PK\x05\x06": add("empty_archive", "correction", None, "The ZIP is empty.") elif b"PK\x03\x04" in raw[:1024 * 1024]: add("prepended_data", "security", None, "The ZIP starts with data that is not part of the archive.") else: add("not_a_zip", "correction", None, "This file is not a ZIP archive.") return None, None try: zf = zipfile.ZipFile(path) except zipfile.BadZipFile: add("truncated_zip", "correction", None, "The ZIP appears to be incomplete or damaged.") return None, None infos = zf.infolist() comment = zf.comment.decode("utf-8", errors="replace") if zf.comment else None # Per-entry comments are metadata too (scanned with the archive comment); unknown extra fields can carry data. entry_comments = [i.comment.decode("utf-8", errors="replace") for i in zf.infolist() if i.comment] if entry_comments: comment = "\n".join([c for c in [comment or ""] + entry_comments if c.strip()]) or None unknown_extra = 0 for i in zf.infolist(): ex, x = i.extra or b"", 0 while x + 4 <= len(ex): hid, size = struct.unpack(" MAX_ENTRIES: add("too_many_entries", "correction", None, "More than %d entries. Include fewer, representative files." % MAX_ENTRIES) names, items, total, unsafe = set(), [], 0, False # Overlapping or hidden data between entries (crafted archives). spans = [] for info in infos: off = info.header_offset if off + 30 > len(raw) or raw[off:off + 4] != b"PK\x03\x04": add("inconsistent_structure", "security", None, "The ZIP's internal directory does not match its contents.") return None, comment n, e = struct.unpack(" a_end: add("hidden_data", "security", None, "The ZIP hides data between its files.") unsafe = True break for info in infos: name = info.filename.replace("\\", "/") upath = _extra_unicode_path(info.extra or b"") if upath is not None and upath.replace("\\", "/") != name: add("header_mismatch", "security", None, "An entry carries a second, different file name.") unsafe = True if name in names: add("duplicate_path", "security", name, "The ZIP lists the same path twice.") unsafe = True names.add(name) if name.startswith("/") or re.match(r"^[A-Za-z]:", name): add("absolute_path", "security", None, "An entry uses an absolute path.") unsafe = True continue if ".." in name.split("/"): add("path_traversal", "security", None, "An entry points outside the folder.") unsafe = True continue mode = (info.external_attr >> 16) & 0xFFFF if info.create_system == 3 and stat.S_ISLNK(mode): add("symlink", "security", name, "A symbolic link is not allowed in a sample.") unsafe = True continue if info.create_system == 3 and mode and not (stat.S_ISREG(mode) or stat.S_ISDIR(mode)): add("special_file", "security", name, "A device, pipe or other special file is not allowed in a sample.") unsafe = True continue if info.is_dir(): continue if info.flag_bits & 0x1: add("encrypted", "correction", name, "This entry is password-protected. Create the ZIP without a password.") items.append((name, None)) continue if info.file_size > max_entry: add("too_large", "correction", name, "This entry is larger than samples allow.") items.append((name, None)) continue if info.file_size > RATIO_FLOOR and info.compress_size and info.file_size / float(info.compress_size) > MAX_RATIO: add("compression_ratio", "security", name, "This entry expands abnormally (a decompression-bomb pattern).") unsafe = True continue total += info.file_size if total > max_total: add("too_large", "correction", None, "The ZIP expands to more than samples allow.") items.append((name, None)) continue try: with zf.open(info) as fh: content = fh.read(info.file_size + 1) if len(content) != info.file_size: add("inflates_beyond_declared", "security", name, "An entry expands to a different size than declared.") unsafe = True continue except zipfile.BadZipFile as e: add("crc_mismatch" if "CRC" in str(e) else "malformed_entry", "correction", name, "An entry is damaged. Create the ZIP again.") items.append((name, None)) continue except NotImplementedError: add("unsupported_compression", "correction", name, "An entry uses a compression method samples do not accept. Use a standard ZIP tool.") items.append((name, None)) continue items.append((name, content)) if unsafe: return None, comment return items, comment # ───────────────────────────────────────────────────────────── the seller's own never-include terms def load_terms(path): """One term per line (company, customer, product or person names, project codenames). Never printed or sent.""" terms = [] with open(path, "r", encoding="utf-8") as fh: for line in fh: t = line.strip() if t and not t.startswith("#") and len(t) >= 3: terms.append(t) return [re.compile(r"(? -?\d+ \+0000$") BUCKETS = ((0, 0), (1, 1), (2, 5), (6, 10), (11, 25), (26, 50), (51, 100), (101, 250), (251, 500), (501, 1000), (1001, 5000)) def bucket(n): for lo, hi in BUCKETS: if lo <= n <= hi: return str(lo) if lo == hi else "%d-%d" % (lo, hi) return "5001+" BUCKET_LABELS = tuple(bucket(lo) for lo, _ in BUCKETS) + ("5001+",) class GitError(Exception): pass SAFE_GIT = ["-c", "core.fsmonitor=false", "-c", "core.hooksPath=" + os.devnull, "-c", "protocol.allow=never"] for _proto in ("file", "git", "ssh", "http", "https", "ext", "fd"): SAFE_GIT += ["-c", "protocol.%s.allow=never" % _proto] SAFE_GIT += ["-c", "credential.helper=", "-c", "core.sshCommand=false", "-c", "diff.external=", "-c", "log.showSignature=false", "-c", "core.pager=cat", "-c", "gc.auto=0", "-c", "maintenance.auto=false", "-c", "core.attributesFile=" + os.devnull, "-c", "core.askPass=", "-c", "init.templateDir=", "-c", "core.quotePath=false", "-c", "transfer.fsckObjects=true", "-c", "init.defaultBranch=main", "--no-pager", "--literal-pathspecs", "--no-replace-objects"] def is_inside(child, parent): """True when `child` is `parent` or inside it (compares real files, so letter case cannot bypass it).""" if not parent or not os.path.exists(parent): return False cur = os.path.realpath(child) while True: if os.path.exists(cur): try: if os.path.samefile(cur, parent): return True except OSError: pass nxt = os.path.dirname(cur) if nxt == cur: return False cur = nxt class Git(object): """The system git, run with every repository-controlled program disabled, from an empty private HOME.""" def __init__(self, forbid_inside=()): exe = shutil.which("git") if not exe: raise SystemExit("Git was not found on this computer. Install Git 2.20 or later and try again.") exe = os.path.realpath(exe) for d in forbid_inside: if d and is_inside(exe, d): raise SystemExit("Refusing to run a git program found inside the repository or the sample folder.") self.exe = exe self.home = tempfile.mkdtemp(prefix="monument-git-") env = {"PATH": os.environ.get("PATH", ""), "HOME": self.home, "XDG_CONFIG_HOME": self.home, "LC_ALL": "C", "LANG": "C", "GIT_CONFIG_NOSYSTEM": "1", "GIT_CONFIG_GLOBAL": os.devnull, "GIT_TERMINAL_PROMPT": "0", "GIT_OPTIONAL_LOCKS": "0", "GIT_NO_LAZY_FETCH": "1", "GIT_PAGER": "cat", "GIT_ALLOW_PROTOCOL": "none", "GIT_ASKPASS": "", "SSH_ASKPASS": "", "GIT_LITERAL_PATHSPECS": "1", "GIT_NO_REPLACE_OBJECTS": "1", "GIT_ATTR_NOSYSTEM": "1"} for k in ("SYSTEMROOT", "SystemRoot", "TMP", "TEMP", "TMPDIR", "COMSPEC"): if k in os.environ: env[k] = os.environ[k] self.env = env def run(self, repo, args, data=None, check=True): cmd = [self.exe] + SAFE_GIT + (["-C", repo] if repo else []) + list(args) p = subprocess.run(cmd, input=data, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=self.env, cwd=self.home, timeout=1800) if check and p.returncode != 0: raise GitError("git %s failed (exit %d)" % (args[0], p.returncode)) return p def out(self, repo, args): return self.run(repo, args).stdout def close(self): shutil.rmtree(self.home, ignore_errors=True) class CatFile(object): """`git cat-file --batch` (no filters, no textconv): raw objects only.""" def __init__(self, git, repo): self.p = subprocess.Popen([git.exe] + SAFE_GIT + ["-C", repo, "cat-file", "--batch"], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, env=git.env, cwd=git.home) def _read(self, n): parts, left = [], n while left > 0: chunk = self.p.stdout.read(min(left, 1 << 20)) if not chunk: break parts.append(chunk) left -= len(chunk) return b"".join(parts) def get(self, oid, limit): self.p.stdin.write(oid.encode("ascii") + b"\n") self.p.stdin.flush() parts = self.p.stdout.readline().split() if len(parts) < 3: return None, None, 0 typ, size = parts[1].decode("ascii"), int(parts[2]) if size > limit: self._read(size + 1) return typ, None, size data = self._read(size) self.p.stdout.read(1) return typ, data, size def close(self): try: self.p.stdin.close() self.p.wait(timeout=120) except Exception: self.p.kill() def open_source_repo(git, path): path = os.path.realpath(path) if not os.path.isdir(path): raise SystemExit("Repository folder not found.") gd = git.run(path, ["rev-parse", "--absolute-git-dir"], check=False) if gd.returncode != 0: raise SystemExit("This folder is not a Git repository (or Git refused it, for example because it belongs to another user).") git_dir = gd.stdout.decode("utf-8", "replace").strip() tp = git.run(path, ["rev-parse", "--show-toplevel"], check=False) top = tp.stdout.decode("utf-8", "replace").strip() if tp.returncode == 0 and tp.stdout.strip() else None if git.out(path, ["rev-parse", "--is-shallow-repository"]).strip() == b"true": raise SystemExit("This is a shallow clone: its history is incomplete. Use a full clone.") partial = git.run(path, ["config", "--get", "extensions.partialclone"], check=False).stdout.strip() promisor = git.run(path, ["config", "--get-regexp", r"^remote\..*\.promisor$"], check=False).stdout if partial or b"true" in promisor: raise SystemExit("This is a partial clone (objects are fetched on demand). Use a full clone.") return path, git_dir, top def repo_fingerprint(git, repo, git_dir): """Refs plus the state files of the repository: compared before and after an export.""" h = hashlib.sha256(git.out(repo, ["for-each-ref", "--format=%(refname) %(objectname)"])) for name in ("HEAD", "index", "packed-refs", "config", "FETCH_HEAD", "ORIG_HEAD", "shallow"): p = os.path.join(git_dir, name) if os.path.isfile(p): st = os.stat(p) h.update(("%s %d %d" % (name, st.st_size, int(getattr(st, "st_mtime_ns", st.st_mtime * 1e9)))).encode()) with open(p, "rb") as fh: h.update(hashlib.sha256(fh.read()).digest()) return h.hexdigest() IDENT = re.compile(r"^(.*?) <([^>]*)> (-?\d+) ([+-]\d{4})$") def parse_commit(raw): head, _, msg = raw.partition(b"\n\n") hdr = {} for line in head.split(b"\n"): if line.startswith(b" "): continue key, _, val = line.partition(b" ") hdr.setdefault(key.decode("latin-1"), []).append(val) def ident(key): v = hdr.get(key, [b""])[0].decode("utf-8", "replace") m = IDENT.match(v) return (m.group(1).strip(), m.group(2).strip(), int(m.group(3))) if m else ("", "", 0) return {"headers": sorted(hdr), "parents": [p.decode("ascii", "replace") for p in hdr.get("parent", [])], "author": ident("author"), "committer": ident("committer"), "message": msg.decode("utf-8", "replace")} def word_rx(s): return re.compile(r"(?= 5 and local.lower() not in GENERIC_LOCAL and re.search(r"[A-Za-z]", local): pats.append((word_rx(local), "contributor-%d" % n)) d = domain.lower() if d and d not in PUBLIC_MAIL and not d.endswith(".noreply.github.com") and not d.endswith(PSEUDO_DOMAIN) and "." in d: pats.append((word_rx(domain), "company.example.invalid")) nm = (name or "").strip() if len(nm) >= 4 and " " in nm: pats.append((word_rx(nm), "Contributor %d" % n)) for part in nm.split(): if len(part) >= 6 and part[0].isupper(): pats.append((word_rx(part), "Contributor %d" % n)) elif len(nm) >= 5 and nm.lower() not in GENERIC_LOCAL: pats.append((word_rx(nm), "contributor-%d" % n)) return pats def local_patterns(extra_paths=()): """Exporter-local metadata (home folder, repository path, user and host names) that must never leave.""" pats = [] for p in [os.path.expanduser("~")] + list(extra_paths): if p and len(p) > 3: pats.append((re.compile(re.escape(p), re.I), "/path/to")) try: user = getpass.getuser() except Exception: user = "" if user and len(user) >= 4 and user.lower() not in GENERIC_LOCAL: pats.append((word_rx(user), "user")) host = platform.node().split(".")[0] if host and len(host) >= 4 and re.search(r"[A-Za-z]", host) and host.lower() not in ("localhost",): pats.append((word_rx(host), "host")) return pats def decode_token(token): if re.match(r"^[0-9a-fA-F]+$", token) and len(token) % 2 == 0: try: d = bytes.fromhex(token).decode("latin-1") except ValueError: return None return d if re.fullmatch(r"[\x20-\x7e\s]{8,}", d) else None t = token.replace("-", "+").replace("_", "/").rstrip("=") if len(t) % 4 == 1: t = t[:-1] t += "=" * ((4 - len(t) % 4) % 4) try: d = base64.b64decode(t).decode("utf-8", "replace") except Exception: return None return d if len(d) >= 8 and PRINTABLE.fullmatch(d) else None class Sanitizer(object): """Replace every detected value with a placeholder. Detections are collected first and overlapping ones merged, so a term inside a password cannot leave the rest of the password behind. Values seen are kept in memory only.""" def __init__(self, k, terms, identity_patterns, local_pats): self.k = k self.terms = terms self.identity = identity_patterns self.local = local_pats self.counts = dict.fromkeys(REDACTION_CATEGORIES, 0) self.where = {} # path → kinds of values replaced in some version of it (for the owner's private notes only) self.emails = {} self.canaries = set() def _email(self, value): key = value.lower() if key not in self.emails: self.emails[key] = "person-%d@%s" % (len(self.emails) + 1, PSEUDO_DOMAIN) return self.emails[key] def _config_secret(self, text): k = self.k for m in k.sensitive_keys.finditer(text): v = re.sub(r"[;)\]]+$", "", m.group(2) or "") if k.sensitive_keys.groups >= 2 else "" if v and not k.placeholders.search(v) and not re.match(r"^\d{1,5}$", v): return True return False def spans(self, text): k = self.k out = [] def add(s, e, cat, repl): if e > s: out.append((s, e, cat, repl)) for m in PEM_BLOCK.finditer(text): add(m.start(), m.end(), "credential", "REDACTED_KEY_BLOCK") for i, rx in enumerate(self.terms): for m in rx.finditer(text): add(m.start(), m.end(), "term", "REDACTED_TERM_%d" % (i + 1)) for rx, repl in self.identity: for m in rx.finditer(text): add(m.start(), m.end(), "identity", repl) for rx, repl in self.local: for m in rx.finditer(text): add(m.start(), m.end(), "local", repl) for _, rx in k.secrets: for m in rx.finditer(text): add(m.start(), m.end(), "credential", "REDACTED_CREDENTIAL") if k.sensitive_keys.groups >= 2: for m in k.sensitive_keys.finditer(text): if m.group(2): v = re.sub(r"[;)\]]+$", "", m.group(2)) if not k.placeholders.search(v) and not re.match(r"^\d{1,5}$", v): add(m.start(2), m.start(2) + len(v), "credential", "REDACTED") lim = k.rules["limits"] for m in re.finditer(r"[A-Za-z0-9+/_-]{%d,}={0,2}" % lim["encodedMinLength"], text): dec = decode_token(m.group(0)) if dec and (any(rx.search(dec) for _, rx in k.secrets) or self._config_secret(dec)): add(m.start(), m.end(), "credential", "REDACTED_ENCODED") for d, rx in k.regulated: accept = luhn if d.get("luhn") else (iban_valid if d["id"] == "iban" else None) for m in rx.finditer(text): if accept is None or accept(m.group(0)): add(m.start(), m.end(), "regulated", "REDACTED_ID") for d, rx in k.pii: for m in rx.finditer(text): v = m.group(0) if d["id"] == "email": if PLACEHOLDER_VIEW.fullmatch(v): continue add(m.start(), m.end(), "personal", self._email(v)) elif d["id"] == "phone": add(m.start(), m.end(), "personal", "+00 000 000 0000") elif d["id"] == "street_address": add(m.start(), m.end(), "personal", "[address]") elif d["id"] == "date_of_birth": add(m.start(), m.end(), "personal", "[date of birth]") repl_infra = {"internal_hostname": "internal.example.invalid", "private_ip": "192.0.2.10", "cloud_account": "REDACTED_ACCOUNT", "internal_tool_url": "https://tools.example.invalid"} for d, rx in k.infra: for m in rx.finditer(text): add(m.start(), m.end(), "infrastructure", repl_infra.get(d["id"], "REDACTED_INTERNAL")) for rx in k.names: for m in rx.finditer(text): if not PLACEHOLDER_VIEW.search(m.group(0)): add(m.start(), m.end(), "personal", "[person]") return out def sanitize(self, text): merged = [] for s, e, cat, repl in sorted(self.spans(text), key=lambda x: (x[0], -x[1])): if merged and s < merged[-1][1]: ps, pe, pcat, prepl = merged[-1] if PRIORITY[cat] < PRIORITY[pcat]: pcat, prepl = cat, repl merged[-1] = (ps, max(pe, e), pcat, prepl) else: merged.append((s, e, cat, repl)) parts, last = [], 0 for s, e, cat, repl in merged: parts.append(text[last:s]) parts.append(repl) last = e self.counts[cat] += 1 val = text[s:e] if cat in ("credential", "regulated", "identity", "term") and len(val) >= 8: self.canaries.add(val) parts.append(text[last:]) return "".join(parts) def scan_view(text): """The text as the re-check sees it: Monument's own placeholders removed so they are not mistaken for data.""" return PLACEHOLDER_VIEW.sub(" ", text) # The code manifest's own identifiers (file SHA-256s, the sample id, commit and tree ids) are random hex: scanned as # text they can look like phone numbers or other identifiers. They are masked before the privacy scan (both here and on # Monument's server); every path and every other value in the manifest is still scanned. CODE_MANIFEST_IDS = re.compile(r"\b(?:[0-9a-f]{64}|[0-9a-f]{40}|[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})\b") def mask_code_manifest_ids(text): return CODE_MANIFEST_IDS.sub("ID", text) # README.md, HISTORY_REPORT.md, TESTS.md, SAMPLE_MANIFEST.json and repository.bundle: the files the ZIP holds besides source/. CODE_TOP_FILE_COUNT = 5 def code_scan_limits(): """How much text Monument's server reads per code sample, and how many files it accepts in the ZIP (as served, the server's own policy values).""" if not CODE_SCAN_LIMITS_JSON.startswith("__MONUMENT"): lim = json.loads(CODE_SCAN_LIMITS_JSON) return {"total": int(lim["scanBytesTotal"]), "per_entry": int(lim["scanBytesPerEntry"]), "max_files": int(lim.get("maxEntries", 2000))} return {"total": 64 * 1024 * 1024, "per_entry": 2 * 1024 * 1024, "max_files": 2000} def too_many_files_message(files, limit): return ("The sample would have %d files; Monument accepts at most %d per code sample (counting the %d files this tool adds) " "and refuses a larger ZIP at upload. Choose fewer folders with --include (code-plan shows the files per folder) and " "run code-export again." % (files, limit, CODE_TOP_FILE_COUNT)) def server_units(text): """The length Monument's server measures for one text: UTF-16 code units (JavaScript string length) after the same placeholder removal and normalization this tool applies.""" return len(normalize_for_scan(scan_view(text)).encode("utf-16-le")) // 2 def mib(n): # Labelled MB like Monument's page (both count 1,048,576 bytes to the MB). return "%.1f MB" % (n / 1048576.0) if n >= 1048576 else "%d KB" % max(1, js_round(n / 1024.0)) def fit_recent(per_commit, docs, limit): """per_commit: oldest first, (message units, {oid: units}) for the whole tree of each commit. The largest number of most recent commits whose distinct file versions, messages and documents fit the limit (0 if not even one does).""" seen, total, best = set(), docs, 0 for n, (msg, blobs) in enumerate(reversed(per_commit), 1): total += msg + sum(u for o, u in blobs.items() if o not in seen) seen.update(blobs) if total > limit: break best = n return best def scan_limit_message(total, limit, best, per_entry_over): if per_entry_over: how = "%d text(s) are larger than %s each, which Monument cannot read in full; leave those files out." % (per_entry_over, mib(code_scan_limits()["per_entry"])) elif best: how = "keep only the most recent %d commit(s) (code-export --max-commits %d), or include fewer folders." % (best, best) else: how = "include fewer folders or files: the latest version of the selected files alone is over the limit." return ("Monument fully inspects at most %s of text per code sample: every version of every file in the sample's history, the " "commit messages and this tool's documents. This sample has about %s, so Monument could not inspect all of it and would hold it " "(it could never be cleared or shared). Make a smaller, representative sample: %s" % (mib(limit), mib(total), how)) def path_exclusion(path, k, terms, identity): segs = path.split("/") if re.search(r"[\x00-\x1f\x7f]", path) or re.search(u"[‪-‮⁦-⁩]", path): return "unsafe_name" if ".git" in segs or segs[-1] in (".gitmodules", ".gitattributes"): return "git_metadata" if is_secret_like(path): return "credential_file" if any(s.startswith(".") for s in segs) and not HIDDEN_ALLOW.search(path): return "hidden_file" if DEPENDENCY_DIR.search(path): return "dependency_or_build" if EXECUTABLE_EXT.search(path): return "executable" if ARCHIVE_EXT.search(path): return "archive" np_ = normalize_for_scan(path) if count(k.pii[0][1], np_) + count(k.pii[1][1], np_) or any(rx.search(np_) for rx, _ in identity): return "identifying_name" if any(rx.search(np_) for rx in terms): return "blocked_term_name" if any(rx.search(np_) for _, rx in k.secrets): return "credential_name" return None def process_blob(data, size, path, sanitizer, k): """(sanitized bytes, review codes) or (None, exclusion reason).""" if data is None: return None, "too_large" if size > MAX_CODE_FILE else "unreadable_text" if data.startswith(LFS_POINTER): return None, "lfs_object" if BINARY_EXT.search(path) or b"\x00" in data[:8192]: return None, "binary" text = decode_strict(data) if text is None: return None, "unreadable_text" before = dict(sanitizer.counts) new = sanitizer.sanitize(normalize_for_scan(text) if INVISIBLE.search(text) else text) kinds = [c for c in REDACTION_CATEGORIES if sanitizer.counts[c] > before[c]] if kinds: sanitizer.where.setdefault(path, set()).update(kinds) _, findings, _ = assess_privacy(sanitizer.k, [{"path": path, "bytes": len(new), "area": "other", "kind": "text", "text": scan_view(new), "structured": structured_kind(path)}], None, None, budget_total=CODE_SCAN_TOTAL, budget_entry=MAX_CODE_FILE * 4) sev = set(f["severity"] for f in findings) if "sensitive" in sev: return None, "residual_sensitive" if sev & {"incomplete", "unsupported"}: return None, "unreadable_text" return (new.encode("utf-8"), sorted(set(f["code"] for f in findings if f["severity"] == "review"))), None def sanitize_message(msg, policy, i, nfiles, sanitizer, k): if policy == "generic": return "Commit %d: changes to %d file(s)\n" % (i + 1, nfiles) m = TRAILER.sub("", msg) m = re.sub(r"https?://[^\s)\]>\"']+", "[link]", m) # Branch, fork and pull-request names can carry customer or project names: keep the wording, drop the names. m = re.sub(r"(Merge (?:pull request|branch|remote-tracking branch|tag))\s+(?:#\d+\s+)?(?:'[^'\n]*'|\"[^\"\n]*\"|\S+)((?:\s+(?:of|from|into)\s+(?:'[^'\n]*'|\S+))*)", lambda mm: mm.group(1) + " [branch]", m) m = sanitizer.sanitize(normalize_for_scan(m)) m = re.sub(r"\n{3,}", "\n\n", m).strip()[:1000].strip() or "Update" _, findings, _ = assess_privacy(k, [{"path": "(commit message)", "bytes": len(m), "area": "other", "kind": "text", "text": scan_view(m), "structured": None}], None, None) if any(f["severity"] in ("sensitive", "review") for f in findings): return "Commit %d: changes to %d file(s)\n" % (i + 1, nfiles) return m + "\n" def stamp(policy, original, i): if policy == "exact": return int(original) t = time.gmtime(int(original)) if policy == "day": return calendar.timegm((t.tm_year, t.tm_mon, t.tm_mday, 12, 0, 0, 0, 0, 0)) + i if policy == "month": return calendar.timegm((t.tm_year, t.tm_mon, 1, 0, 0, 0, 0, 0, 0)) + i * 60 return BASE_EPOCH + i * 3600 def git_blob_id(content): return hashlib.sha1(b"blob %d\x00" % len(content) + content).hexdigest() def fi_path(path): return '"' + path.replace("\\", "\\\\").replace('"', '\\"') + '"' def write_history(git, work, commits): """Rebuild the sample history in a fresh repository with git fast-import (nothing from the original .git).""" git.run(None, ["init", "-q", work]) git.run(work, ["symbolic-ref", "HEAD", "refs/heads/main"]) chunks, marks, nxt = [], {}, 1 for c in commits: for _, (mode, content) in sorted(c["tree"].items()): if content not in marks: marks[content] = nxt chunks.append(b"blob\nmark :%d\ndata %d\n" % (nxt, len(content)) + content + b"\n") nxt += 1 prev = None for c in commits: mark = nxt nxt += 1 msg = c["message"].encode("utf-8") head = [b"commit refs/heads/main", b"mark :%d" % mark, ("author Contributor %d %d +0000" % (c["author"], c["author"], PSEUDO_DOMAIN, c["time"])).encode(), ("committer Contributor %d %d +0000" % (c["committer"], c["committer"], PSEUDO_DOMAIN, c["time"])).encode(), b"data %d" % len(msg)] body = b"\n".join(head) + b"\n" + msg + b"\n" if prev is not None: body += b"from :%d\n" % prev body += b"deleteall\n" for path, (mode, content) in sorted(c["tree"].items()): body += ("M %s :%d %s\n" % (mode, marks[content], fi_path(path))).encode("utf-8") chunks.append(body + b"\n") prev = mark chunks.append(b"done\n") git.run(work, ["fast-import", "--quiet", "--done"], data=b"".join(chunks)) def language_of(path): return LANGUAGE_OF.get(os.path.splitext(path)[1].lower()) def collect_identities(git, repo, head, identities): raw = git.out(repo, ["rev-list", "--format=%an%x00%ae%x00%cn%x00%ce", head]).decode("utf-8", "replace") for line in raw.split("\n"): if not line or line.startswith("commit "): continue parts = line.split("\x00") if len(parts) == 4: identities.number(parts[0], parts[1]) identities.number(parts[2], parts[3]) def code_export(args): rules = load_rules(args.rules) k = Compiled(rules) terms = load_terms(args.terms) if args.terms else [] out_root = os.path.realpath(args.out) max_commits = max(1, min(int(args.max_commits or MAX_COMMITS_DEFAULT), MAX_COMMITS_LIMIT)) git = Git(forbid_inside=(os.path.realpath(args.repo), out_root)) work = tempfile.mkdtemp(prefix="monument-code-sample-") try: repo, git_dir, top = open_source_repo(git, args.repo) base = top or git_dir if is_inside(out_root, base) or is_inside(out_root, git_dir): raise SystemExit("Choose an output folder outside the repository.") candidate = os.path.join(out_root, CODE_DIR) if os.path.exists(candidate) and os.listdir(candidate): raise SystemExit("The output folder already contains a sample. Choose a new, empty folder.") before = repo_fingerprint(git, repo, git_dir) rev = args.branch or "HEAD" hp = git.run(repo, ["rev-parse", "--verify", "--quiet", "--end-of-options", rev + "^{commit}"], check=False) if hp.returncode != 0: raise SystemExit("The branch or commit to sample was not found.") head = hp.stdout.decode("ascii").strip() includes = [] for inc in args.include or []: inc = inc.replace("\\", "/").strip().strip("/") while inc.startswith("./"): inc = inc[2:] if not inc or inc.startswith("/") or ".." in inc.split("/") or re.match(r"^[A-Za-z]:", inc): raise SystemExit("Each --include must be a path inside the repository.") if not git.out(repo, ["ls-tree", "--full-tree", "--name-only", head, "--", inc]).strip(): raise SystemExit("A path given with --include does not exist in the selected branch.") includes.append(inc) if not includes: raise SystemExit("Choose at least one folder or file to include (--include PATH).") listed = git.out(repo, ["rev-list", "--first-parent", "--reverse", head, "--"] + includes).decode("ascii").split() if not listed: raise SystemExit("No commits touch the selected paths.") selected = listed[-max_commits:] cat = CatFile(git, repo) identities = Identities() metas = [] for c in selected: _, raw, _ = cat.get(c, 1 << 20) info = parse_commit(raw or b"") metas.append((c, info, identities.number(info["author"][0], info["author"][1]), identities.number(info["committer"][0], info["committer"][1]))) collect_identities(git, repo, head, identities) id_pats = identities.patterns() sanitizer = Sanitizer(k, terms, id_pats, local_patterns([repo, base, git_dir, out_root])) # Pass 1: classify every version of every path; a path left out once is left out of every commit. versions, excluded, blobs, review = [], {}, {}, {} for c, info, a, cm in metas: entries = [] for ent in git.out(repo, ["ls-tree", "-r", "-z", "--full-tree", c, "--"] + includes).split(b"\x00"): if not ent: continue meta, _, raw_path = ent.partition(b"\t") mode, typ, oid = meta.decode("ascii").split() try: path = raw_path.decode("utf-8") except UnicodeDecodeError: excluded["(undecodable name)"] = "unsafe_name" continue if typ == "commit": excluded[path] = "submodule" elif mode == "120000": excluded[path] = "symlink" else: reason = path_exclusion(path, k, terms, id_pats) if reason: excluded[path] = reason else: if (oid, path) not in blobs: _, data, size = cat.get(oid, MAX_CODE_FILE) blobs[(oid, path)] = process_blob(data, size, path, sanitizer, k) res, why = blobs[(oid, path)] if why: excluded.setdefault(path, why) else: if res[1]: review.setdefault(path, set()).update(res[1]) entries.append((path, "100755" if mode == "100755" else "100644", oid)) versions.append(entries) cat.close() # Pass 2: build trees without excluded paths; skip commits that change nothing in the sample. commits, prev = [], None for (c, info, a, cm), entries in zip(metas, versions): tree = {} for path, mode, oid in entries: if path not in excluded: tree[path] = (mode, blobs[(oid, path)][0][0]) sig = tuple(sorted((p, m, git_blob_id(b)) for p, (m, b) in tree.items())) if not tree or sig == prev: continue prev = sig i = len(commits) commits.append({"author": a, "committer": cm, "time": stamp(args.timestamps, info["committer"][2], i), "message": sanitize_message(info["message"], args.messages, i, len(tree), sanitizer, k), "tree": tree}) if not commits: raise SystemExit("No safe, meaningful sample could be produced: every selected file was left out (see the reasons with --json). Choose other paths, or describe the code instead (metadata only).") final = commits[-1]["tree"] source_files = [p for p in final if language_of(p)] if not source_files: raise SystemExit("No safe, meaningful sample could be produced: no source code file survived the checks. Choose other paths, or describe the code instead (metadata only).") # Renumber contributors densely by appearance in the sample (pseudonyms only). order = {} for c in commits: for key in ("author", "committer"): order.setdefault(c[key], len(order) + 1) c[key] = order[c[key]] repo_work = os.path.join(work, "history") write_history(git, repo_work, commits) git.run(repo_work, ["fsck", "--full", "--strict", "--no-dangling", "--no-progress"]) new_ids = git.out(repo_work, ["rev-list", "--reverse", "main"]).decode("ascii").split() if len(new_ids) != len(commits): raise SystemExit("The rebuilt history does not match the selection; nothing was written.") for cid, c in zip(new_ids, commits): got = set() for ent in git.out(repo_work, ["ls-tree", "-r", "-z", cid]).split(b"\x00"): if ent: meta, _, p = ent.partition(b"\t") mode, _, oid = meta.decode("ascii").split() got.add((p.decode("utf-8"), mode, oid)) if got != set((p, m, git_blob_id(b)) for p, (m, b) in c["tree"].items()): raise SystemExit("The rebuilt history does not match the sanitized files; nothing was written.") os.makedirs(candidate, exist_ok=True) bundle = os.path.join(candidate, "repository.bundle") git.run(repo_work, ["bundle", "create", bundle, "HEAD", "main"]) git.run(repo_work, ["bundle", "verify", bundle]) for path, (mode, content) in final.items(): dest = os.path.join(candidate, "source", *path.split("/")) os.makedirs(os.path.dirname(dest), exist_ok=True) with open(dest, "wb") as fh: fh.write(content) tests = sorted(p for p in final if TEST_PATH.search(p)) langs = sorted(set(language_of(p) for p in source_files)) excl_counts = {} for reason in excluded.values(): excl_counts[reason] = excl_counts.get(reason, 0) + 1 contributors = len(order) head_new = new_ids[-1] head_tree = git.out(repo_work, ["rev-parse", "main^{tree}"]).decode("ascii").strip() write_code_docs(candidate, commits, contributors, args, excl_counts, sanitizer.counts, tests, langs) manifest = {"format": CODE_MANIFEST_FORMAT, "format_version": CODE_FORMAT_VERSION, "tool": TOOL, "tool_version": CHECKER_VERSION, "rules_version": rules["rulesVersion"], "sample_id": str(uuid.uuid4()), "created_on": datetime.date.today().isoformat(), "history": {"branch": "main", "commits": len(commits), "contributors": contributors, "timestamps": args.timestamps, "messages": args.messages, "rewritten": True, "merges_linearized": True, "head_commit": head_new, "head_tree": head_tree}, "content": {"files": len(final), "source_files": len(source_files), "test_files": len(tests), "languages": langs}, "exclusions": excl_counts, "redactions": dict(sanitizer.counts), "local_checks": None, "approval": None, "files": []} write_code_manifest(candidate, manifest) after = repo_fingerprint(git, repo, git_dir) if after != before: raise SystemExit("The repository changed while the sample was being made. Nothing should be shared; run the export again when nothing else is using the repository.") result = verify_code_candidate(candidate, git, k, terms, id_pats, sanitizer.canaries, local_patterns([repo, base, git_dir, out_root])) result["checks"]["source-unchanged"] = "pass" # Record the checks only the export can run (it alone saw the original values) for the certificate. manifest["local_checks"] = {"status": result["status"], "checks": result["checks"]} with open(os.path.join(candidate, CODE_MANIFEST), "w", encoding="utf-8") as fh: fh.write(json.dumps(manifest, indent=2, sort_keys=True) + "\n") write_private_review(out_root, excluded, review, result, k, sanitizer.where) result["summary"] = {"commits": len(commits), "contributors": contributors, "files": len(final), "source_files": len(source_files), "test_files": len(tests), "languages": langs, "excluded": excl_counts, "redactions": dict(sanitizer.counts), "review_files": len(review)} return result finally: shutil.rmtree(work, ignore_errors=True) git.close() def write_code_docs(candidate, commits, contributors, args, excl, redactions, tests, langs): w = lambda name, text: open(os.path.join(candidate, name), "w", encoding="utf-8").write(text) w("README.md", "# Code evaluation sample (sanitized)\n\nPrepared on the owner's computer with the Monument sample tool %s.\n\n" "- `repository.bundle` — the sample's Git history: one branch, `main`, %d commit(s). Restore it with `git clone repository.bundle sample-repo`.\n" "- `source/` — the files of the latest commit, for reading without Git.\n" "- `HISTORY_REPORT.md` — how the history was rewritten and what was left out.\n" "- `TESTS.md` — the tests included.\n" "- `SAMPLE_MANIFEST.json` — size and SHA-256 of every file here.\n\n" "This is a limited evaluation sample, not the full repository. Sensitive values were replaced by placeholders (for example REDACTED_CREDENTIAL). Automated checks cannot prove that nothing confidential remains.\n" % (CHECKER_VERSION, len(commits))) lines = ["# History report", "", "- Commits included: %d (first-parent history of the selected paths, oldest first)." % len(commits), "- Contributors: %d, shown as pseudonyms (\"Contributor 1\", …). Numbers follow order of appearance; they are not derived from real identities." % contributors, "- Commit ids differ from the original repository; merges were linearised; tags, notes, signatures and other branches are not included.", "- Timestamps: %s." % {"none": "not disclosed (a synthetic sequence keeps the order only)", "month": "rounded to the month", "day": "rounded to the day", "exact": "exact, in UTC"}[args.timestamps], "- Commit messages: %s." % ("sanitized (sign-off and co-author lines, links and branch names removed; messages that still looked sensitive were replaced)" if args.messages == "sanitized" else "replaced by generic text"), "", "## Left out", ""] lines += ["- %s: %d path(s)" % (r.replace("_", " "), n) for r, n in sorted(excl.items())] or ["- nothing"] lines += ["", "## Values replaced", ""] + ["- %s: %d" % (c, n) for c, n in sorted(redactions.items()) if n] + ["", "## Limitations", ""] + ["- " + l for l in CODE_LIMITATIONS] w("HISTORY_REPORT.md", "\n".join(lines) + "\n") t = ["# Tests", "", "- Test files included: %d." % len(tests), "- Languages: %s." % (", ".join(langs) or "none detected"), ""] t += ["- `%s`" % p for p in tests[:200]] t += ["", "The tool never runs code. The owner may report whether the tests pass on a clone of `repository.bundle`; that report is the owner's statement."] w("TESTS.md", "\n".join(t) + "\n") def write_code_manifest(candidate, manifest): files = [] for dirpath, dirnames, filenames in os.walk(candidate): dirnames.sort() for name in sorted(filenames): full = os.path.join(dirpath, name) rel = os.path.relpath(full, candidate).replace(os.sep, "/") if rel == CODE_MANIFEST: continue with open(full, "rb") as fh: data = fh.read() files.append({"path": rel, "sha256": hashlib.sha256(data).hexdigest(), "bytes": len(data)}) manifest["files"] = files with open(os.path.join(candidate, CODE_MANIFEST), "w", encoding="utf-8") as fh: fh.write(json.dumps(manifest, indent=2, sort_keys=True) + "\n") def write_private_review(out_root, excluded, review, result, k=None, replaced=None): """Owner-only notes (paths and reasons, never values — paths with emails, phones or credentials are masked). Never part of the sample or the certificate.""" mp = (lambda p: mask_path_values(k, p)) if k else (lambda p: p) d = os.path.join(out_root, "private-review") os.makedirs(d, exist_ok=True) try: os.chmod(d, 0o700) except OSError: pass lines = ["# Private review notes (keep on this computer; never share)", "", "How to review: read these lists with the owner, together with the sample's README.md and HISTORY_REPORT.md.", "If anything should not be shared, run code-export again into a NEW, empty folder with fewer folders (--include)", "or fewer commits (--max-commits). Package (code-package) only after the owner approves exactly this sample.", "", "## Left out of the sample", ""] lines += ["- `%s`: %s" % (mp(p), r.replace("_", " ")) for p, r in sorted(excluded.items())] or ["- nothing"] lines += ["", "## Values replaced with placeholders (in the current file or an earlier version of it)", ""] lines += ["- `%s`: %s" % (mp(p), ", ".join(c for c in REDACTION_CATEGORIES if c in kinds)) for p, kinds in sorted((replaced or {}).items())] or ["- nothing"] if any("credential" in kinds for kinds in (replaced or {}).values()) or any(r in ("credential_file", "credential_name") for r in excluded.values()): lines += ["", "ROTATE real credentials: a credential replaced here, or in a credential file left out above, is still in the", "original repository's history and must be treated as exposed — removing it from the sample does not revoke it."] lines += ["", "## Kept, but needs your judgement (REVIEW_REQUIRED)", ""] lines += ["- `%s`: %s" % (mp(p), ", ".join(sorted(c))) for p, c in sorted(review.items())] or ["- nothing"] lines += ["", "## Re-check of the sample", "", "Status: %s" % result["status"]] lines += ["- [%s] %s%s" % (f["severity"], f["code"], " — " + f["path"] if f["path"] else "") for f in result["findings"]] p = os.path.join(d, "REVIEW.md") with open(p, "w", encoding="utf-8") as fh: fh.write("\n".join(lines) + "\n") try: os.chmod(p, 0o600) except OSError: pass def verify_code_candidate(folder, git, k, terms, identity_patterns=None, canaries=None, local_pats=None): """Independent re-check of a sample folder: layout, manifest, history (bundle, fsck, headers, every historical blob and message), snapshot = HEAD, and leak checks. Reads only the sample; trusts nothing it contains.""" findings, checks = [], {} add = lambda code, sev, path, msg: findings.append({"code": code, "severity": sev, "path": path, "message": msg}) def check(name, ok, skip=False): checks[name] = "skip" if skip else ("pass" if ok and checks.get(name) != "fail" else "fail") files, total = {}, 0 layout_ok, internals_ok = True, True for dirpath, dirnames, filenames in os.walk(folder, followlinks=False): for dn in list(dirnames): if os.path.islink(os.path.join(dirpath, dn)): add("symlink", "security", None, "The sample contains a symbolic link.") layout_ok = False dirnames.remove(dn) for name in filenames: full = os.path.join(dirpath, name) rel = os.path.relpath(full, folder).replace(os.sep, "/") st = os.lstat(full) if not stat.S_ISREG(st.st_mode): add("special_file", "security", rel, "The sample contains a link or special file.") layout_ok = False continue segs = rel.split("/") if ".git" in segs or segs[-1] in (".gitmodules", ".gitattributes"): add("git_internals", "sensitive", rel, "The sample contains Git internals; only repository.bundle may carry history.") internals_ok = False top = segs[0] if not ((len(segs) == 1 and top in CODE_TOP_FILES) or (top == "source" and len(segs) > 1)): add("unexpected_file", "correction", rel, "This file is not part of a code sample.") layout_ok = False elif top == "source" and any(s.startswith(".") for s in segs[1:]) and not HIDDEN_ALLOW.search("/".join(segs[1:])): add("hidden_file", "correction", rel, "Hidden files do not belong in a code sample.") layout_ok = False total += st.st_size with open(full, "rb") as fh: files[rel] = fh.read() for req in CODE_TOP_FILES: if req not in files: add("missing_required_file", "correction", req, "%s is missing. Create the sample again with the Monument tool." % req) layout_ok = False if total > MAX_CODE_SAMPLE_BYTES: add("sample_too_large", "correction", None, "The sample is larger than %d MB. Select fewer paths or commits." % (MAX_CODE_SAMPLE_BYTES // (1024 * 1024))) layout_ok = False check("archive-layout", layout_ok) check("no-git-internals", internals_ok) manifest = None try: manifest = json.loads(files.get(CODE_MANIFEST, b"").decode("utf-8")) except Exception: add("manifest_invalid", "correction", CODE_MANIFEST, "SAMPLE_MANIFEST.json is not valid JSON.") man_ok = isinstance(manifest, dict) and manifest.get("format") == CODE_MANIFEST_FORMAT and manifest.get("format_version") == CODE_FORMAT_VERSION and isinstance(manifest.get("files"), list) if man_ok: listed = {} for f in manifest["files"]: if not isinstance(f, dict) or not isinstance(f.get("path"), str) or f["path"] in listed: man_ok = False break listed[f["path"]] = f actual = {p: d for p, d in files.items() if p != CODE_MANIFEST} if set(listed) != set(actual): man_ok = False for p, d in actual.items(): f = listed.get(p) if not f or f.get("sha256") != hashlib.sha256(d).hexdigest() or f.get("bytes") != len(d): man_ok = False if not man_ok: add("manifest_mismatch", "correction", CODE_MANIFEST, "SAMPLE_MANIFEST.json does not match the files in the sample.") check("manifest-integrity", man_ok) texts = [] # (label, text) for leak checks # What Monument's server will read (its scan limit): each distinct file version once, the commit messages, the documents. server_seen, per_commit, entry_units, doc_units = {}, [], [], 0 hist_ok, meta_ok, snap_ok = False, False, False head_files = {} if "repository.bundle" in files: tmp = tempfile.mkdtemp(prefix="monument-verify-") try: repo_v = os.path.join(tmp, "r") git.run(None, ["init", "-q", "--bare", repo_v]) bpath = os.path.join(tmp, "repository.bundle") with open(bpath, "wb") as fh: fh.write(files["repository.bundle"]) heads = git.run(repo_v, ["bundle", "list-heads", bpath], check=False).stdout.decode("ascii", "replace").split("\n") heads = sorted(h for h in heads if h.strip()) ok_verify = git.run(repo_v, ["bundle", "verify", bpath], check=False).returncode == 0 names = sorted(h.split(" ")[-1] for h in heads) shas = set(h.split(" ")[0] for h in heads) if ok_verify and names in (["refs/heads/main"], ["HEAD", "refs/heads/main"]) and len(shas) == 1: un = git.run(repo_v, ["bundle", "unbundle", bpath], check=False) sha = shas.pop() if un.returncode == 0 and re.match(r"^[0-9a-f]{40}$", sha): git.run(repo_v, ["update-ref", "refs/heads/main", sha]) hist_ok = git.run(repo_v, ["fsck", "--full", "--strict", "--no-dangling", "--no-progress"], check=False).returncode == 0 if not hist_ok: add("history_invalid", "security", "repository.bundle", "The history bundle is incomplete, has extra or missing refs, or fails Git's integrity check.") if hist_ok: ids = git.out(repo_v, ["rev-list", "--reverse", "main"]).decode("ascii").split() cat = CatFile(git, repo_v) meta_ok = True blobs = set() core = [] for i, cid in enumerate(ids): per_commit.append([0, {}]) _, raw, _ = cat.get(cid, 1 << 20) raw = raw or b"" head, _, msg = raw.partition(b"\n\n") keys = [] for line in head.split(b"\n"): key, _, val = line.partition(b" ") keys.append(key) if key in (b"author", b"committer") and not GIT_IDENT_PSEUDO.match(val.decode("utf-8", "replace")): meta_ok = False if set(keys) - {b"tree", b"parent", b"author", b"committer"} or keys.count(b"parent") > 1: meta_ok = False text = msg.decode("utf-8", "replace") texts.append(("(commit message)", text)) texts.append(("(commit header)", head.decode("utf-8", "replace"))) core.append({"path": "(commit message %d)" % (i + 1), "bytes": len(text), "area": "other", "kind": "text", "text": scan_view(text), "structured": None}) per_commit[-1][0] = server_units(text) entry_units.append(per_commit[-1][0]) for ent in git.out(repo_v, ["ls-tree", "-r", "-z", cid]).split(b"\x00"): if not ent: continue m, _, p = ent.partition(b"\t") mode, typ, oid = m.decode("ascii").split() path = p.decode("utf-8", "replace") if typ != "blob" or mode not in ("100644", "100755"): meta_ok = False continue if path_exclusion(path, k, terms or [], identity_patterns or []) in ("hidden_file", "credential_file", "git_metadata", "identifying_name", "blocked_term_name", "credential_name", "unsafe_name"): add("excluded_path_in_history", "sensitive", path, "The history contains a file that must never be part of a sample.") if oid in server_seen: per_commit[-1][1][oid] = server_seen[oid] if (oid, path) in blobs: continue blobs.add((oid, path)) _, data, size = cat.get(oid, MAX_CODE_FILE) if data is None or b"\x00" in data[:8192]: add("unsupported_content", "unsupported", path, "A file in the history is binary or too large to check.") continue txt = decode_strict(data) if txt is None: add("unreadable_content", "unsupported", path, "A file in the history is not valid text.") continue if oid not in server_seen: # Monument reads each distinct file version once, however many paths or commits carry it. server_seen[oid] = server_units(txt) entry_units.append(server_seen[oid]) per_commit[-1][1][oid] = server_seen[oid] texts.append((path, txt)) core.append({"path": path, "bytes": len(txt), "area": "other", "kind": "text", "text": scan_view(txt), "structured": structured_kind(path)}) if i == len(ids) - 1: pass if i == len(ids) - 1: for ent in git.out(repo_v, ["ls-tree", "-r", "-z", cid]).split(b"\x00"): if ent: m, _, p = ent.partition(b"\t") head_files[p.decode("utf-8", "replace")] = m.decode("ascii").split()[2] cat.close() if git.out(repo_v, ["for-each-ref", "--format=%(refname)"]).decode("ascii").split() != ["refs/heads/main"]: meta_ok = False _, pf, _ = assess_privacy(k, core, None, None, budget_total=CODE_SCAN_TOTAL, budget_entry=MAX_CODE_FILE * 4) findings.extend(pf) if not meta_ok: add("git_metadata", "sensitive", "repository.bundle", "Commits carry headers or identities other than the sample's pseudonyms (signatures, encodings, real names), extra refs, links or submodules.") finally: shutil.rmtree(tmp, ignore_errors=True) check("history-reconstructable", hist_ok) check("git-metadata", meta_ok) snap = {p[len("source/"):]: git_blob_id(d) for p, d in files.items() if p.startswith("source/")} snap_ok = hist_ok and snap == head_files if hist_ok and not snap_ok: add("snapshot_mismatch", "correction", "source/", "The source/ snapshot does not equal the latest commit of the history.") check("snapshot-matches-history", snap_ok) for p, d in files.items(): if p.startswith("source/") or p == "repository.bundle": continue txt = decode_strict(d) or "" texts.append((p, txt)) scanned = mask_code_manifest_ids(txt) if p == CODE_MANIFEST else txt doc_units += server_units(scanned) entry_units.append(server_units(scanned)) # The sample's own documents are scanned as text (the manifest's keys are the tool's vocabulary, not records). _, pf, _ = assess_privacy(k, [{"path": p, "bytes": len(scanned), "area": "other", "kind": "text", "text": scan_view(scanned), "structured": None}], None, None) findings.extend(pf) snap_core = [] for p, d in files.items(): if p.startswith("source/"): txt = decode_strict(d) if txt is None or b"\x00" in d[:8192]: add("unsupported_content", "unsupported", p, "A snapshot file is binary or not valid text.") continue texts.append((p, txt)) snap_core.append({"path": p, "bytes": len(txt), "area": "other", "kind": "text", "text": scan_view(txt), "structured": structured_kind(p)}) if snap_core: _, pf, _ = assess_privacy(k, snap_core, None, None, budget_total=CODE_SCAN_TOTAL, budget_entry=MAX_CODE_FILE * 4) findings.extend(pf) def leak(name, patterns, code, message): if patterns is None: check(name, True, skip=True) return hit = 0 for _, t in texts: for rx in patterns: if rx.search(t): hit += 1 break if hit: add(code, "sensitive", None, message % hit) check(name, not hit) leak("identity-leak-check", [rx for rx, _ in identity_patterns] if identity_patterns is not None else None, "identity_leak", "An original contributor name or address survives in %d place(s) of the sample.") leak("blocked-term-check", terms if terms else None, "blocked_term", "A never-include term survives in %d place(s) of the sample.") leak("local-metadata-check", [rx for rx, _ in local_pats] if local_pats is not None else None, "local_metadata", "Exporter-local metadata (home folder, user or host name, repository path) survives in %d place(s) of the sample.") leak("known-value-canaries", [re.compile(re.escape(c)) for c in canaries] if canaries else None, "redacted_value_survives", "A value that was redacted still appears in %d place(s) of the sample.") tests = [p for p in head_files if TEST_PATH.search(p)] check("tests-included", bool(tests), skip=not tests) # Monument's limit: a sample it cannot read in full is held there and can never be cleared, so it is not packaged here. lim = code_scan_limits() server_total = doc_units + sum(m for m, _ in per_commit) + sum(server_seen.values()) over_entry = len([u for u in entry_units if u > lim["per_entry"]]) best = fit_recent([(m, b) for m, b in per_commit], doc_units, lim["total"]) if hist_ok else 0 inspection = {"text_bytes": server_total, "limit_bytes": lim["total"], "fits": bool(hist_ok) and server_total <= lim["total"] and not over_entry, "max_recent_commits_that_fit": best, "commits": len(per_commit)} if hist_ok and not inspection["fits"]: add("monument_scan_limit", "incomplete", None, scan_limit_message(server_total, lim["total"], best, over_entry)) # Monument's file limit: a larger ZIP is refused at upload, so it is not packaged here either. inspection["files"] = len(head_files) + CODE_TOP_FILE_COUNT inspection["max_files"] = lim["max_files"] inspection["files_fit"] = inspection["files"] <= lim["max_files"] if not inspection["files_fit"]: add("too_many_files", "correction", None, too_many_files_message(inspection["files"], lim["max_files"])) content_ok = not any(f["severity"] in ("sensitive", "unsupported", "incomplete") for f in findings) check("content-rescan", content_ok) structural = [n for n in ("archive-layout", "no-git-internals", "manifest-integrity", "history-reconstructable", "git-metadata", "snapshot-matches-history") if checks.get(n) == "fail"] status = status_of(findings) if structural and status in ("CHECKS_PASSED", "REVIEW_REQUIRED"): status = "SCAN_INCOMPLETE" # A ZIP Monument would refuse at upload is corrected here first: it is never packaged. if not inspection["files_fit"] and status in ("CHECKS_PASSED", "REVIEW_REQUIRED"): status = "NEEDS_CORRECTION" for f in findings: if f["path"]: f["path"] = mask_path_values(k, f["path"]) return {"status": status, "checks": checks, "findings": dedupe(findings), "structural_failures": structural, "manifest": manifest if man_ok else None, "stats": {"commits": len(ids) if hist_ok else 0, "files": len(head_files), "test_files": len(tests), "monument_inspection": inspection}} def build_code_certificate(manifest, rules, status, checks, zip_sha, manifest_sha, approval): h, c = manifest["history"], manifest["content"] return { "format": CODE_CERT_FORMAT, "format_version": CODE_FORMAT_VERSION, "tool": TOOL, "tool_version": CHECKER_VERSION, "rules_version": rules["rulesVersion"], "certificate_id": str(uuid.uuid4()), "created_on": datetime.date.today().isoformat(), "result": status, "result_meaning": CODE_RESULT_MEANING, "custody": CODE_CUSTODY, "binding": {"sample_sha256": zip_sha, "manifest_sha256": manifest_sha}, "sample": {"commits": bucket(h["commits"]), "contributors": bucket(h["contributors"]), "files": bucket(c["files"]), "test_files": bucket(c["test_files"]), "languages": [l for l in c["languages"] if l in LANGUAGES][:12], "history": {"rewritten": True, "original_commit_ids_disclosed": False, "merges_linearized": True, "single_branch": True, "timestamps": h["timestamps"], "messages": h["messages"]}}, "exclusions": {r: bucket(n) for r, n in sorted(manifest["exclusions"].items()) if r in EXCLUSION_REASONS}, "redactions": {r: bucket(n) for r, n in sorted(manifest["redactions"].items()) if r in REDACTION_CATEGORIES}, "checks": [{"id": i, "result": checks.get(i, "skip")} for i in CODE_CHECK_IDS], "seller_approval": {"approved_by_role": approval["approved_by_role"], "approved_on": approval["approved_on"], "review_items_acknowledged": bucket(approval["review_items_acknowledged"]), "tests_result": approval["tests_result"], "custody": {"monument_may_hold": approval["custody"]["monument_may_hold"], "purposes": approval["custody"]["purposes"], "sharing": approval["custody"]["sharing"], "custody_days": approval["custody"]["custody_days"]}}, "limitations": list(CODE_LIMITATIONS), } def validate_code_certificate(cert): """Strict allowlist schema (mirrored by the server): anything unexpected is refused rather than forwarded.""" def keys(d, allowed): if not isinstance(d, dict) or set(d) != set(allowed): raise ValueError("certificate: unexpected or missing fields") keys(cert, ("format", "format_version", "tool", "tool_version", "rules_version", "certificate_id", "created_on", "result", "result_meaning", "custody", "binding", "sample", "exclusions", "redactions", "checks", "seller_approval", "limitations")) ok = (cert["format"] == CODE_CERT_FORMAT and cert["format_version"] == CODE_FORMAT_VERSION and cert["tool"] == TOOL and re.match(r"^\d+\.\d+\.\d+$", cert["tool_version"]) and re.match(r"^[a-z0-9-]{1,64}$", cert["rules_version"]) and re.match(r"^[0-9a-f-]{36}$", cert["certificate_id"]) and re.match(r"^\d{4}-\d{2}-\d{2}$", cert["created_on"]) and cert["result"] in ("CHECKS_PASSED", "REVIEW_REQUIRED") and cert["result_meaning"] == CODE_RESULT_MEANING and cert["custody"] == CODE_CUSTODY and tuple(cert["limitations"]) == CODE_LIMITATIONS) keys(cert["binding"], ("sample_sha256", "manifest_sha256")) ok = ok and all(re.match(r"^[0-9a-f]{64}$", v) for v in cert["binding"].values()) s = cert["sample"] keys(s, ("commits", "contributors", "files", "test_files", "languages", "history")) ok = ok and all(s[x] in BUCKET_LABELS for x in ("commits", "contributors", "files", "test_files")) and isinstance(s["languages"], list) and all(l in LANGUAGES for l in s["languages"]) keys(s["history"], ("rewritten", "original_commit_ids_disclosed", "merges_linearized", "single_branch", "timestamps", "messages")) ok = ok and s["history"]["rewritten"] is True and s["history"]["original_commit_ids_disclosed"] is False and s["history"]["timestamps"] in TIMESTAMP_POLICIES and s["history"]["messages"] in MESSAGE_POLICIES ok = ok and all(k in EXCLUSION_REASONS and v in BUCKET_LABELS for k, v in cert["exclusions"].items()) and all(k in REDACTION_CATEGORIES and v in BUCKET_LABELS for k, v in cert["redactions"].items()) ok = ok and isinstance(cert["checks"], list) and [c.get("id") for c in cert["checks"]] == list(CODE_CHECK_IDS) and all(set(c) == {"id", "result"} and c["result"] in ("pass", "fail", "skip") for c in cert["checks"]) a = cert["seller_approval"] keys(a, ("approved_by_role", "approved_on", "review_items_acknowledged", "tests_result", "custody")) keys(a["custody"], ("monument_may_hold", "purposes", "sharing", "custody_days")) ok = ok and a["approved_by_role"] in APPROVER_ROLES and re.match(r"^\d{4}-\d{2}-\d{2}$", a["approved_on"]) and a["review_items_acknowledged"] in BUCKET_LABELS and a["tests_result"] in TESTS_RESULTS ok = ok and isinstance(a["custody"]["monument_may_hold"], bool) and isinstance(a["custody"]["purposes"], list) and a["custody"]["purposes"] and all(p in EVALUATION_PURPOSES for p in a["custody"]["purposes"]) ok = ok and a["custody"]["sharing"] in SHARING_SCOPES and isinstance(a["custody"]["custody_days"], int) and 7 <= a["custody"]["custody_days"] <= 365 if not ok: raise ValueError("certificate: a value is outside the allowed vocabulary") def write_code_zip(candidate, zip_path): entries = [] for dirpath, dirnames, filenames in os.walk(candidate): dirnames.sort() for name in sorted(filenames): full = os.path.join(dirpath, name) entries.append((CODE_DIR + "/" + os.path.relpath(full, candidate).replace(os.sep, "/"), full)) with zipfile.ZipFile(zip_path, "x", compression=zipfile.ZIP_DEFLATED) as zf: for arc, full in entries: info = zipfile.ZipInfo(arc, date_time=(1980, 1, 1, 0, 0, 0)) info.external_attr = (0o100644 & 0xFFFF) << 16 info.compress_type = zipfile.ZIP_STORED if arc.endswith(".bundle") else zipfile.ZIP_DEFLATED info.create_system = 3 with open(full, "rb") as fh: zf.writestr(info, fh.read()) def code_verify_zip(zip_path, git, k, terms, identity_patterns, local_pats): findings = [] add = lambda code, sev, path, msg: findings.append({"code": code, "severity": sev, "path": path, "message": msg}) items, comment = gather_zip(zip_path, add, max_entry=MAX_CODE_SAMPLE_BYTES, max_total=MAX_CODE_SAMPLE_BYTES + 16 * 1024 * 1024) if comment: add("archive_metadata", "review", None, "The ZIP carries a comment. Create it without one.") if items is None: return {"status": "SCAN_INCOMPLETE", "checks": {"zip-reverified": "fail"}, "findings": findings, "structural_failures": ["zip"], "manifest": None, "stats": {}} tmp = tempfile.mkdtemp(prefix="monument-zipcheck-") try: for name, content in items: if not name.startswith(CODE_DIR + "/") or content is None: add("unexpected_file", "correction", name, "Everything in a code sample ZIP must be inside monument-code-sample/ and readable.") continue dest = os.path.join(tmp, *name.split("/")) os.makedirs(os.path.dirname(dest), exist_ok=True) with open(dest, "wb") as fh: fh.write(content) res = verify_code_candidate(os.path.join(tmp, CODE_DIR), git, k, terms, identity_patterns, None, local_pats) finally: shutil.rmtree(tmp, ignore_errors=True) res["findings"] = dedupe(findings + res["findings"]) status = status_of(res["findings"]) if res["structural_failures"] or any(f["severity"] in ("security", "correction") for f in findings): status = "SCAN_INCOMPLETE" if status in ("CHECKS_PASSED", "REVIEW_REQUIRED") else status res["status"] = status res["checks"]["zip-reverified"] = "pass" if status in ("CHECKS_PASSED", "REVIEW_REQUIRED") and not res["structural_failures"] else "fail" return res def code_package(args): rules = load_rules(args.rules) k = Compiled(rules) terms = load_terms(args.terms) if args.terms else [] out_root = os.path.realpath(args.out) candidate = os.path.join(out_root, CODE_DIR) if not os.path.isdir(candidate): raise SystemExit("No sample folder found. Run code-export first.") if args.approved_by_role not in APPROVER_ROLES: raise SystemExit("--approved-by-role must be one of: %s" % ", ".join(APPROVER_ROLES)) if args.confirm != APPROVAL_PHRASE: raise SystemExit('The owner must approve the exact sample: pass --confirm "%s" only after they reviewed it.' % APPROVAL_PHRASE) purposes = [p.strip() for p in (args.purposes or "").split(",") if p.strip()] if not purposes or any(p not in EVALUATION_PURPOSES for p in purposes): raise SystemExit("--purposes must list one or more of: %s" % ", ".join(EVALUATION_PURPOSES)) if args.sharing not in SHARING_SCOPES: raise SystemExit("--sharing must be one of: %s" % ", ".join(SHARING_SCOPES)) if not 7 <= int(args.custody_days) <= 365: raise SystemExit("--custody-days must be between 7 and 365.") zip_path = os.path.join(out_root, CODE_DIR + ".zip") if os.path.exists(zip_path): raise SystemExit("monument-code-sample.zip already exists. Remove it only if you mean to replace it.") git = Git(forbid_inside=(out_root,)) try: id_pats = None if args.repo: repo, git_dir, top = open_source_repo(git, args.repo) ids = Identities() collect_identities(git, repo, args.branch or "HEAD", ids) id_pats = ids.patterns() locals_ = local_patterns([os.path.realpath(args.repo)] if args.repo else []) res = verify_code_candidate(candidate, git, k, terms, id_pats, None, locals_) if res["structural_failures"] or res["status"] in ("SENSITIVE_DATA_FOUND", "SCAN_INCOMPLETE", "UNSUPPORTED_CONTENT", "NEEDS_CORRECTION"): res["refused"] = "The sample did not pass its checks, so no ZIP was created. Fix the findings (run code-export again) and repeat." return res review_n = len([f for f in res["findings"] if f["severity"] == "review"]) if res["status"] == "REVIEW_REQUIRED" and not args.accept_review: res["refused"] = "Some items need the owner's judgement. Show them the private review notes; package only if they accept them (--accept-review)." return res manifest_path = os.path.join(candidate, CODE_MANIFEST) with open(manifest_path, "r", encoding="utf-8") as fh: manifest = json.load(fh) approval = {"approved_by_role": args.approved_by_role, "approved_on": datetime.date.today().isoformat(), "review_items_acknowledged": review_n, "tests_result": args.tests_result, "custody": {"monument_may_hold": bool(args.allow_monument_custody), "purposes": purposes, "sharing": args.sharing, "custody_days": int(args.custody_days)}} prior = (manifest.get("local_checks") or {}).get("checks") or {} manifest["approval"] = approval manifest["local_checks"] = {"status": res["status"], "checks": res["checks"]} write_code_manifest(candidate, manifest) write_code_zip(candidate, zip_path) zres = code_verify_zip(zip_path, git, k, terms, id_pats, locals_) if zres["status"] not in ("CHECKS_PASSED", "REVIEW_REQUIRED") or zres["structural_failures"]: os.remove(zip_path) zres["refused"] = "The final ZIP failed its own inspection and was deleted. Nothing should be shared." return zres with open(zip_path, "rb") as fh: zip_sha = hashlib.sha256(fh.read()).hexdigest() with open(manifest_path, "rb") as fh: manifest_sha = hashlib.sha256(fh.read()).hexdigest() checks = dict(res["checks"], **{"zip-reverified": zres["checks"]["zip-reverified"]}) for name in ("source-unchanged", "known-value-canaries"): checks[name] = "pass" if prior.get(name) == "pass" else "skip" cert = build_code_certificate(manifest, rules, zres["status"], checks, zip_sha, manifest_sha, approval) validate_code_certificate(cert) cert_text = json.dumps(cert, indent=2, sort_keys=True) + "\n" leak_pats = ([rx for rx, _ in id_pats] if id_pats else []) + list(terms) + [rx for rx, _ in locals_] if any(rx.search(cert_text) for rx in leak_pats): os.remove(zip_path) raise SystemExit("The certificate would carry identifying text; nothing was written.") cert_path = os.path.join(out_root, "monument-certificate.json") with open(cert_path, "w", encoding="utf-8") as fh: fh.write(cert_text) zres["package"] = {"zip": zip_path, "bytes": os.path.getsize(zip_path), "sha256": zip_sha, "certificate": cert_path} return zres finally: git.close() def plan_estimate(git, repo, head, args): """Before anything is built: about how much text Monument would inspect for the most recent --max-commits commits of the selected paths (sizes only, never contents). Text-like files the export would keep are counted once per distinct version; the export itself measures exactly and refuses to package a sample that does not fit.""" lim = code_scan_limits() includes = [i.replace("\\", "/").strip().strip("/") for i in (getattr(args, "include", None) or []) if i.strip()] n = max(1, min(int(getattr(args, "max_commits", None) or MAX_COMMITS_DEFAULT), MAX_COMMITS_LIMIT)) listed = git.out(repo, ["rev-list", "--first-parent", "--reverse", head, "--"] + includes).decode("ascii").split()[-n:] per_commit, sizes, folder_of = [], {}, {} head_files = 0 for c in listed: tree = {} for ent in git.out(repo, ["ls-tree", "-r", "-l", "-z", "--full-tree", c, "--"] + includes).split(b"\x00"): if not ent: continue meta, _, raw = ent.partition(b"\t") parts = meta.decode("ascii").split() if len(parts) < 4 or parts[1] != "blob" or parts[0] == "120000" or not parts[3].isdigit(): continue path = raw.decode("utf-8", "replace") size = int(parts[3]) if size > MAX_CODE_FILE or BINARY_EXT.search(path) or (any(x.startswith(".") for x in path.split("/")) and not HIDDEN_ALLOW.search(path)): continue # the export leaves these out sizes[parts[2]] = size folder_of.setdefault(parts[2], path.split("/")[0] if "/" in path else "(root files)") tree[parts[2]] = size if c == listed[-1]: head_files += 1 per_commit.append((512, tree)) # a sanitized commit message is short # README, HISTORY_REPORT and TESTS, and the manifest (about 200 characters per file listed), generously. docs = 8192 + 256 * len(per_commit[-1][1] if per_commit else {}) total = docs + sum(m for m, _ in per_commit) + sum(sizes.values()) folders = {} for oid, size in sizes.items(): folders[folder_of[oid]] = folders.get(folder_of[oid], 0) + size files = head_files + CODE_TOP_FILE_COUNT return {"commits_considered": len(listed), "paths": includes or ["(whole branch)"], "text_bytes": total, "limit_bytes": lim["total"], "fits": total <= lim["total"], "max_recent_commits_that_fit": fit_recent(per_commit, docs, lim["total"]), "folders": folders, "files": files, "max_files": lim["max_files"], "files_fit": files <= lim["max_files"]} def code_plan(args): """A read-only overview of one repository to choose what to sample (paths and counts only; never contents).""" git = Git(forbid_inside=(os.path.realpath(args.repo),)) try: repo, git_dir, top = open_source_repo(git, args.repo) head = git.out(repo, ["rev-parse", "--verify", "--end-of-options", (args.branch or "HEAD") + "^{commit}"]).decode("ascii").strip() commits = int(git.out(repo, ["rev-list", "--count", head]).strip() or 0) ids = Identities() collect_identities(git, repo, head, ids) folders = {} hidden = secretish = submodules = symlinks = 0 for ent in git.out(repo, ["ls-tree", "-r", "-z", "--full-tree", head]).split(b"\x00"): if not ent: continue meta, _, p = ent.partition(b"\t") mode, typ, _ = meta.decode("ascii").split() path = p.decode("utf-8", "replace") if typ == "commit": submodules += 1 continue if mode == "120000": symlinks += 1 if any(s.startswith(".") for s in path.split("/")) and not HIDDEN_ALLOW.search(path): hidden += 1 if is_secret_like(path): secretish += 1 topd = path.split("/")[0] if "/" in path else "(root files)" f = folders.setdefault(topd, {"files": 0, "source": 0, "tests": 0, "languages": set()}) f["files"] += 1 if language_of(path): f["source"] += 1 f["languages"].add(language_of(path)) if TEST_PATH.search(path): f["tests"] += 1 estimate = plan_estimate(git, repo, head, args) return {"commits": commits, "contributors": len(ids.people), "submodules": submodules, "symlinks": symlinks, "hidden_files": hidden, "credential_like_files": secretish, "folders": {k2: {"files": v["files"], "source_files": v["source"], "test_files": v["tests"], "languages": sorted(v["languages"]), "estimated_text_bytes": estimate["folders"].get(k2, 0)} for k2, v in sorted(folders.items())}, "monument_inspection_estimate": {k2: v for k2, v in estimate.items() if k2 != "folders"}} finally: git.close() def print_code_report(rep, out, title): w = out.write w("Monument sample tool %s — %s (offline; Git runs with repository programs disabled)\n\n" % (CHECKER_VERSION, title)) if "summary" in rep: s = rep["summary"] w("Sample: %d commit(s), %d pseudonymous contributor(s), %d file(s) (%d source, %d test) · languages: %s\n" % (s["commits"], s["contributors"], s["files"], s["source_files"], s["test_files"], ", ".join(s["languages"]) or "none")) w("Left out: %s\n" % (", ".join("%s %d" % (r.replace("_", " "), n) for r, n in sorted(s["excluded"].items())) or "nothing")) w("Values replaced: %s\n" % (", ".join("%s %d" % (c, n) for c, n in sorted(s["redactions"].items()) if n) or "none")) if s["redactions"].get("credential") or any(r in ("credential_file", "credential_name") for r in s["excluded"]): w("Credentials: the original repository still holds them — rotate any real one (private-review/REVIEW.md lists the files).\n") w("Files kept that need the owner's judgement: %d (see private-review/REVIEW.md)\n" % s["review_files"]) ins = (rep.get("stats") or {}).get("monument_inspection") if ins: w("Text Monument will inspect: %s of at most %s — %s\n" % (mib(ins["text_bytes"]), mib(ins["limit_bytes"]), "within the limit" if ins["fits"] else "TOO MUCH: Monument could not inspect it all (see monument_scan_limit below)")) if "files" in ins: w("Files in the ZIP: %d of at most %d — %s\n" % (ins["files"], ins["max_files"], "within the limit" if ins["files_fit"] else "TOO MANY: Monument refuses it at upload (see too_many_files below)")) w("\nStatus: %s\n" % rep["status"]) for name in CODE_CHECK_IDS: if name in rep.get("checks", {}): w(" %-26s %s\n" % (name, rep["checks"][name])) order = ["security", "sensitive", "correction", "unsupported", "incomplete", "review", "warning"] for f in sorted(rep.get("findings", []), key=lambda x: order.index(x["severity"]) if x["severity"] in order else 9): w(" [%s] %s%s\n %s\n" % (f["severity"].upper(), f["code"], (" — " + f["path"]) if f["path"] else "", f["message"])) if rep.get("refused"): w("\nNOT PACKAGED: %s\n" % rep["refused"]) elif title == "code-export": if rep["status"] in ("CHECKS_PASSED", "REVIEW_REQUIRED") and not rep.get("structural_failures"): w("\nNext: the owner reviews the sample — README.md and HISTORY_REPORT.md in monument-code-sample/, and private-review/REVIEW.md " "(stays on this computer). %sPackage it with code-package only after the owner approves.\n" % ("The items marked REVIEW need the owner's judgement; " if rep["status"] == "REVIEW_REQUIRED" else "")) else: w("\nNext: do not package this sample. Fix what is listed above (usually fewer folders or fewer commits) and run code-export again into a new, empty folder.\n") elif title == "code-verify" and rep["status"] in ("CHECKS_PASSED", "REVIEW_REQUIRED") and not rep.get("structural_failures"): w("\nNext: upload monument-code-sample.zip (and monument-certificate.json) through your Monument invitation link. Do not email it or upload it anywhere else.\n") if rep.get("package"): p = rep["package"] w("\nCreated: %s\nSize: %d bytes\nSHA-256: %s\nCertificate (metadata only): %s\n" % (p["zip"], p["bytes"], p["sha256"], p["certificate"])) w("Next: submit the ZIP through your secure Monument code-sample invitation. Do not email it or upload it anywhere else.\n") w("\nWhat this does not prove: automated checks cannot establish that nothing confidential remains, and they are not legal clearance.\nNo detected value is printed here.\n") def code_exit(rep): if rep.get("refused") or rep.get("structural_failures"): return STATUS_EXIT.get(rep["status"], 1) or 1 return STATUS_EXIT.get(rep["status"], 1) def print_report(rep, out): w = out.write w("Monument sample checker %s (rules %s) — offline, no network access\n" % (rep["version"], rep["rulesVersion"])) w("Checked: the %s\n" % ("staged folder" if rep["target"] == "folder" else "final ZIP, entry by entry")) w("Your never-include terms: %s\n\n" % ("%d loaded (never printed)" % rep["customTerms"] if rep["customTerms"] else "none given (use --terms FILE)")) w("Result: %s\n" % rep["outcome"]) w("Privacy status: %s\n" % rep["privacyStatus"]) c = rep["coverage"] w("Scanned: %d file(s), %d characters · not checkable: %d · not fully scanned: %d\n" % (c["scannedFiles"], c["scannedBytes"], c["unsupported"], c["incomplete"])) w("Locally check-complete: %s\n\n" % ("yes" if rep["locallyCheckComplete"] else "NO — fix the findings below and run the checker again")) order = ["security", "sensitive", "correction", "unsupported", "incomplete", "review", "warning"] for f in sorted(rep["findings"], key=lambda x: order.index(x["severity"])): w(" [%s] %s%s\n %s\n" % (f["severity"].upper(), f["code"], (" — " + f["path"]) if f["path"] else "", f["message"])) if not rep["findings"]: w(" No findings by these rules within what could be scanned.\n") w("\nWhat this does not prove:\n") for line in rep["limits"]: w(" - %s\n" % line) w("\nNo detected value is printed in this report. Do not paste file contents into an online service to investigate a finding.\n") def hashes(root): files = [] for dirpath, dirnames, filenames in os.walk(root, followlinks=False): dirnames.sort() for name in sorted(filenames): full = os.path.join(dirpath, name) if os.path.islink(full) or not os.path.isfile(full): continue rel = os.path.relpath(full, root).replace(os.sep, "/") with open(full, "rb") as fh: data = fh.read() files.append((rel, hashlib.sha256(data).hexdigest(), len(data))) role = lambda p: ROLE_OF_AREA.get(sample_area(p), "documentation") listed = [{"path": p, "sha256": s, "bytes": b, "role": role(p)} for p, s, b in files if p not in ("sample_manifest.json", "CHECKSUMS.sha256")] sys.stdout.write("sample_manifest.json \"files\" (every file except sample_manifest.json and CHECKSUMS.sha256):\n") sys.stdout.write(json.dumps(listed, indent=2) + "\n\n") sys.stdout.write("CHECKSUMS.sha256 lines (every file except CHECKSUMS.sha256 — recompute after the manifest changes):\n") for p, s, _ in files: if p != "CHECKSUMS.sha256": sys.stdout.write("%s %s\n" % (s, p)) def main(argv=None): ap = argparse.ArgumentParser(prog="monument_sample_check.py", description="Offline Monument sample checker (no network access).") ap.add_argument("--version", action="version", version="%s %s" % (TOOL, CHECKER_VERSION)) sub = ap.add_subparsers(dest="cmd") for name, help_ in (("check-folder", "check the staged monument-sample-output folder"), ("check-zip", "inspect the final ZIP itself, entry by entry"), ("hashes", "print sizes and SHA-256 for the manifest and CHECKSUMS.sha256")): p = sub.add_parser(name, help=help_) p.add_argument("path") if name != "hashes": p.add_argument("--json", action="store_true", help="machine-readable output") p.add_argument("--terms", help="a local text file of terms that must never appear (one per line); keep it outside the package") p.add_argument("--rules", help=argparse.SUPPRESS) cp = sub.add_parser("code-plan", help="overview of one repository to choose what to sample (read-only)") cp.add_argument("--repo", required=True) cp.add_argument("--branch") cp.add_argument("--include", action="append", help="estimate for these folders or files only (repeatable)") cp.add_argument("--max-commits", type=int, default=MAX_COMMITS_DEFAULT, help="estimate for this many recent commits") cp.add_argument("--json", action="store_true") ce = sub.add_parser("code-export", help="build a sanitized Code + Git History + Tests sample from one repository (the repository is only read)") ce.add_argument("--repo", required=True) ce.add_argument("--include", action="append", help="a folder or file to include (repeatable)") ce.add_argument("--branch") ce.add_argument("--max-commits", type=int, default=MAX_COMMITS_DEFAULT) ce.add_argument("--timestamps", choices=TIMESTAMP_POLICIES, default="none") ce.add_argument("--messages", choices=MESSAGE_POLICIES, default="sanitized") ce.add_argument("--out", required=True, help="a new, empty folder outside the repository") ce.add_argument("--terms", help="a local text file of never-include terms (one per line); keep it outside the sample") ce.add_argument("--json", action="store_true") ce.add_argument("--rules", help=argparse.SUPPRESS) pk = sub.add_parser("code-package", help="after the owner approved the sample: create monument-code-sample.zip and its certificate") pk.add_argument("--out", required=True, help="the folder code-export wrote to") pk.add_argument("--repo", help="the original repository (re-checks that no contributor identity survives)") pk.add_argument("--branch") pk.add_argument("--terms") pk.add_argument("--approved-by-role", required=True, choices=APPROVER_ROLES) pk.add_argument("--confirm", required=True) pk.add_argument("--purposes", required=True, help="comma-separated: " + ", ".join(EVALUATION_PURPOSES)) pk.add_argument("--sharing", required=True, choices=SHARING_SCOPES) pk.add_argument("--custody-days", type=int, default=90) pk.add_argument("--allow-monument-custody", action="store_true", help="the owner authorizes Monument to hold this exact sample") pk.add_argument("--tests-result", choices=TESTS_RESULTS, default="not_run") pk.add_argument("--accept-review", action="store_true", help="the owner reviewed and accepts the REVIEW_REQUIRED items") pk.add_argument("--json", action="store_true") pk.add_argument("--rules", help=argparse.SUPPRESS) cv = sub.add_parser("code-verify", help="inspect a monument-code-sample.zip entry by entry, including its whole Git history") cv.add_argument("path") cv.add_argument("--repo", help="the original repository, to re-check contributor identities") cv.add_argument("--terms") cv.add_argument("--json", action="store_true") cv.add_argument("--rules", help=argparse.SUPPRESS) args = ap.parse_args(argv) if args.cmd in ("code-plan", "code-export", "code-package", "code-verify"): return code_main(args) if not args.cmd: ap.print_help() return 2 if args.cmd == "hashes": hashes(args.path) return 0 rules = load_rules(getattr(args, "rules", None)) terms = load_terms(args.terms) if getattr(args, "terms", None) else None rep = run_check(args.path, "folder" if args.cmd == "check-folder" else "zip", rules, terms) if args.json: sys.stdout.write(json.dumps(rep, indent=2) + "\n") else: print_report(rep, sys.stdout) return STATUS_EXIT[rep["outcome"]] def code_main(args): try: if args.cmd == "code-plan": rep = code_plan(args) if args.json: sys.stdout.write(json.dumps(rep, indent=2) + "\n") else: sys.stdout.write("Repository overview (read-only; contents are never printed)\n\nCommits on the branch: %d · contributors: %d · submodules: %d · symbolic links: %d · hidden files: %d · credential-like files: %d\n\n" % (rep["commits"], rep["contributors"], rep["submodules"], rep["symlinks"], rep["hidden_files"], rep["credential_like_files"])) for name, f in rep["folders"].items(): sys.stdout.write(" %-40s %5d files · %5d source · %4d tests · ~%s of text%s\n" % (name[:40], f["files"], f["source_files"], f["test_files"], mib(f["estimated_text_bytes"]), (" · " + ", ".join(f["languages"])) if f["languages"] else "")) e = rep["monument_inspection_estimate"] sys.stdout.write("\nMonument fully inspects at most %s of text per code sample (every version of every file in the history counts).\n" % mib(e["limit_bytes"])) sys.stdout.write("Estimate for %s, the last %d commit(s): about %s — %s\n" % (", ".join(e["paths"]), e["commits_considered"], mib(e["text_bytes"]), "within the limit" if e["fits"] else "OVER THE LIMIT: choose fewer folders, or at most about %d recent commit(s) (--max-commits)" % e["max_recent_commits_that_fit"] if e["max_recent_commits_that_fit"] else "OVER THE LIMIT even for the latest version alone: choose fewer folders or files")) sys.stdout.write("Files in the ZIP: about %d of at most %d — %s\n" % (e["files"], e["max_files"], "within the limit" if e["files_fit"] else "TOO MANY: Monument refuses it at upload; choose fewer folders")) return 0 rules = load_rules(getattr(args, "rules", None)) if args.cmd == "code-export": rep = code_export(args) title = "code-export" elif args.cmd == "code-package": rep = code_package(args) title = "code-package" else: git = Git() try: k = Compiled(rules) terms = load_terms(args.terms) if args.terms else [] id_pats = None if args.repo: repo, _, _ = open_source_repo(git, args.repo) ids = Identities() collect_identities(git, repo, "HEAD", ids) id_pats = ids.patterns() rep = code_verify_zip(args.path, git, k, terms, id_pats, local_patterns()) finally: git.close() title = "code-verify" if args.json: sys.stdout.write(json.dumps(rep, indent=2, default=list) + "\n") else: print_code_report(rep, sys.stdout, title) return code_exit(rep) except GitError as e: sys.stderr.write("Stopped: %s. Nothing was shared or changed.\n" % e) return 40 if __name__ == "__main__": sys.exit(main())