Jesse
349 lines
15 KiB
Python
349 lines
15 KiB
Python
#!/usr/bin/env python3
|
|
"""Safely import an explicit research allowlist into the commercial repository.
|
|
|
|
The importer is deliberately one-way and offline. It never writes semantic-merge
|
|
or protected paths, and defaults to a dry run. A source without usable Git
|
|
metadata is recorded as an auditable snapshot with ``source_commit=unknown``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import difflib
|
|
import fnmatch
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
from datetime import UTC, datetime
|
|
from pathlib import Path, PurePosixPath
|
|
from typing import Any
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
DEFAULT_POLICY = ROOT / "references/cross_project_contract/sync_policy.v2.json"
|
|
EXIT_OK = 0
|
|
EXIT_SEMANTIC_REVIEW = 2
|
|
EXIT_POLICY_REJECTED = 3
|
|
EXIT_VALIDATION_FAILED = 4
|
|
|
|
|
|
class ImportRejected(ValueError):
|
|
"""The requested import violates a direction, path, privacy or repository guard."""
|
|
|
|
|
|
def sha256_bytes(value: bytes) -> str:
|
|
return hashlib.sha256(value).hexdigest()
|
|
|
|
|
|
def sha256_file(path: Path) -> str:
|
|
return sha256_bytes(path.read_bytes())
|
|
|
|
|
|
def _safe_relative(value: str) -> str:
|
|
if not isinstance(value, str) or not value.strip():
|
|
raise ImportRejected("empty_path")
|
|
candidate = PurePosixPath(value.replace("\\", "/"))
|
|
if candidate.is_absolute() or ".." in candidate.parts or "." in candidate.parts:
|
|
raise ImportRejected(f"unsafe_path:{value}")
|
|
normalized = candidate.as_posix()
|
|
if normalized.startswith("/") or "\x00" in normalized:
|
|
raise ImportRejected(f"unsafe_path:{value}")
|
|
return normalized
|
|
|
|
|
|
def _resolved_file(root: Path, relative: str, *, must_exist: bool) -> Path:
|
|
path = root / _safe_relative(relative)
|
|
root_resolved = root.resolve()
|
|
if must_exist and not path.exists():
|
|
return path
|
|
resolved = path.resolve(strict=must_exist)
|
|
if resolved != root_resolved and root_resolved not in resolved.parents:
|
|
raise ImportRejected(f"symlink_escape:{relative}")
|
|
if must_exist and (not path.is_file() or path.is_symlink()):
|
|
raise ImportRejected(f"non_regular_or_symlink:{relative}")
|
|
return path
|
|
|
|
|
|
def _matches(path: str, patterns: list[str]) -> bool:
|
|
lowered = path.casefold()
|
|
return any(fnmatch.fnmatchcase(lowered, pattern.casefold()) for pattern in patterns)
|
|
|
|
|
|
def load_policy(path: Path) -> dict[str, Any]:
|
|
policy = json.loads(path.read_text(encoding="utf-8"))
|
|
if policy.get("schema_version") != 2:
|
|
raise ImportRejected("policy_schema_version_must_be_2")
|
|
if policy.get("direction") != "research_to_commercial_only":
|
|
raise ImportRejected("direction_must_be_research_to_commercial_only")
|
|
if policy.get("reverse_sync") != "forbidden":
|
|
raise ImportRejected("reverse_sync_must_be_forbidden")
|
|
if "commercial_to_research" in json.dumps(policy, sort_keys=True):
|
|
raise ImportRejected("reverse_direction_is_not_expressible")
|
|
modes = policy.get("modes")
|
|
if not isinstance(modes, dict) or set(modes) != {"mirror", "semantic_merge", "protected"}:
|
|
raise ImportRejected("policy_modes_invalid")
|
|
if not isinstance(modes["mirror"], list) or not isinstance(modes["semantic_merge"], list):
|
|
raise ImportRejected("policy_allowlists_invalid")
|
|
protected = [_safe_relative(value) if "*" not in value else value.replace("\\", "/") for value in modes["protected"]]
|
|
for mapping in modes["mirror"]:
|
|
if not isinstance(mapping, dict) or not {"source", "target", "license"} <= set(mapping):
|
|
raise ImportRejected("mirror_mapping_invalid")
|
|
_safe_relative(mapping["source"])
|
|
target = _safe_relative(mapping["target"])
|
|
if _matches(target, protected):
|
|
raise ImportRejected(f"protected_rejected:{target}")
|
|
for value in modes["semantic_merge"]:
|
|
_safe_relative(value)
|
|
return policy
|
|
|
|
|
|
def _git_output(root: Path, *args: str) -> str | None:
|
|
try:
|
|
return subprocess.check_output(
|
|
["git", "-C", str(root), *args], stderr=subprocess.DEVNULL, text=True, timeout=8
|
|
).strip()
|
|
except (OSError, subprocess.SubprocessError):
|
|
return None
|
|
|
|
|
|
def source_identity(source: Path) -> tuple[str, str]:
|
|
inside = _git_output(source, "rev-parse", "--show-toplevel")
|
|
commit = _git_output(source, "rev-parse", "HEAD") if inside and Path(inside).resolve() == source.resolve() else None
|
|
if commit and len(commit) == 40 and all(ch in "0123456789abcdef" for ch in commit.lower()):
|
|
return "git", commit.lower()
|
|
return "snapshot", "unknown"
|
|
|
|
|
|
def _normalized_remote(root: Path) -> str | None:
|
|
remote = _git_output(root, "remote", "get-url", "origin")
|
|
if not remote:
|
|
return None
|
|
value = remote.strip().lower().removesuffix(".git").replace("git@", "ssh://git@")
|
|
return value.rstrip("/")
|
|
|
|
|
|
def validate_roots(source: Path, target: Path) -> None:
|
|
source = source.expanduser().resolve(strict=True)
|
|
target = target.expanduser().resolve(strict=True)
|
|
if not source.is_dir() or not target.is_dir():
|
|
raise ImportRejected("source_and_target_must_be_directories")
|
|
if source == target or source in target.parents or target in source.parents:
|
|
raise ImportRejected("source_and_target_must_not_overlap")
|
|
source_remote = _normalized_remote(source)
|
|
target_remote = _normalized_remote(target)
|
|
if source_remote and target_remote and source_remote == target_remote:
|
|
raise ImportRejected("source_and_target_git_remote_must_differ")
|
|
|
|
|
|
def source_tree_hash(source: Path) -> str:
|
|
digest = hashlib.sha256()
|
|
for path in sorted(source.rglob("*"), key=lambda item: item.relative_to(source).as_posix()):
|
|
relative = path.relative_to(source).as_posix()
|
|
if ".git" in PurePosixPath(relative).parts or path.is_symlink() or not path.is_file():
|
|
continue
|
|
digest.update(relative.encode("utf-8"))
|
|
digest.update(b"\0")
|
|
digest.update(bytes.fromhex(sha256_file(path)))
|
|
digest.update(b"\0")
|
|
return digest.hexdigest()
|
|
|
|
|
|
def _privacy_rejection(path: Path, relative: str, privacy: dict[str, Any]) -> str | None:
|
|
lowered = relative.casefold()
|
|
basenames = {str(value).casefold() for value in privacy.get("forbidden_basenames", [])}
|
|
extensions = [str(value).casefold() for value in privacy.get("forbidden_extensions", [])]
|
|
if path.name.casefold() in basenames or any(lowered.endswith(ext) for ext in extensions):
|
|
return f"sensitive_filename:{relative}"
|
|
data = path.read_bytes()
|
|
if b"\x00" in data[:8192]:
|
|
return f"binary_content_rejected:{relative}"
|
|
text = data.decode("utf-8", errors="replace")
|
|
for marker in privacy.get("content_markers", []):
|
|
if str(marker).casefold() in text.casefold():
|
|
return f"sensitive_content:{relative}"
|
|
return None
|
|
|
|
|
|
def _diff_summary(source: Path | None, target: Path | None) -> str:
|
|
if source is None:
|
|
return "source missing"
|
|
if target is None:
|
|
return "target missing; manual semantic merge required"
|
|
try:
|
|
before = target.read_text(encoding="utf-8").splitlines()
|
|
after = source.read_text(encoding="utf-8").splitlines()
|
|
except UnicodeDecodeError:
|
|
return "binary difference; manual semantic merge required"
|
|
added = removed = 0
|
|
for line in difflib.ndiff(before, after):
|
|
added += line.startswith("+ ")
|
|
removed += line.startswith("- ")
|
|
return f"manual semantic merge required: +{added}/-{removed} lines"
|
|
|
|
|
|
def build_manifest(
|
|
source: Path,
|
|
*,
|
|
policy_path: Path = DEFAULT_POLICY,
|
|
target: Path = ROOT,
|
|
apply: bool = False,
|
|
) -> dict[str, Any]:
|
|
source = source.expanduser().resolve(strict=True)
|
|
target = target.expanduser().resolve(strict=True)
|
|
validate_roots(source, target)
|
|
policy = load_policy(policy_path)
|
|
protected = list(policy["modes"]["protected"])
|
|
privacy = policy.get("privacy") if isinstance(policy.get("privacy"), dict) else {}
|
|
source_mode, source_commit = source_identity(source)
|
|
target_commit = _git_output(target, "rev-parse", "HEAD")
|
|
if not target_commit or len(target_commit) != 40:
|
|
raise ImportRejected("target_base_commit_unavailable")
|
|
|
|
mirror_files: list[dict[str, Any]] = []
|
|
privacy_rejections: list[str] = []
|
|
pending_writes: list[tuple[Path, bytes, dict[str, Any]]] = []
|
|
for mapping in policy["modes"]["mirror"]:
|
|
source_rel = _safe_relative(mapping["source"])
|
|
target_rel = _safe_relative(mapping["target"])
|
|
if _matches(target_rel, protected):
|
|
raise ImportRejected(f"protected_rejected:{target_rel}")
|
|
source_file = _resolved_file(source, source_rel, must_exist=True)
|
|
if not source_file.exists():
|
|
raise ImportRejected(f"mirror_source_missing:{source_rel}")
|
|
target_file = _resolved_file(target, target_rel, must_exist=False)
|
|
rejection = _privacy_rejection(source_file, source_rel, privacy)
|
|
if rejection:
|
|
privacy_rejections.append(rejection)
|
|
continue
|
|
data = source_file.read_bytes()
|
|
source_hash = sha256_bytes(data)
|
|
before_hash = sha256_file(target_file) if target_file.exists() and target_file.is_file() else None
|
|
status = "unchanged" if source_hash == before_hash else "update" if before_hash else "new"
|
|
row = {
|
|
"source": source_rel,
|
|
"target": target_rel,
|
|
"status": status,
|
|
"source_sha256": source_hash,
|
|
"target_sha256_before": before_hash,
|
|
"target_sha256_after": source_hash,
|
|
"license": str(mapping["license"]),
|
|
}
|
|
mirror_files.append(row)
|
|
if status != "unchanged":
|
|
pending_writes.append((target_file, data, row))
|
|
|
|
if privacy_rejections:
|
|
raise ImportRejected(";".join(privacy_rejections))
|
|
|
|
semantic_files: list[dict[str, Any]] = []
|
|
for relative_value in policy["modes"]["semantic_merge"]:
|
|
relative = _safe_relative(relative_value)
|
|
source_file = _resolved_file(source, relative, must_exist=False)
|
|
target_file = _resolved_file(target, relative, must_exist=False)
|
|
source_exists = source_file.exists() and source_file.is_file() and not source_file.is_symlink()
|
|
target_exists = target_file.exists() and target_file.is_file() and not target_file.is_symlink()
|
|
source_hash = sha256_file(source_file) if source_exists else None
|
|
target_hash = sha256_file(target_file) if target_exists else None
|
|
if not source_exists:
|
|
status = "source_missing"
|
|
elif not target_exists:
|
|
status = "target_missing"
|
|
elif source_hash == target_hash:
|
|
status = "unchanged"
|
|
else:
|
|
status = "review_required"
|
|
semantic_files.append({
|
|
"path": relative,
|
|
"status": status,
|
|
"source_sha256": source_hash,
|
|
"target_sha256": target_hash,
|
|
"diff_summary": _diff_summary(source_file if source_exists else None, target_file if target_exists else None),
|
|
})
|
|
|
|
if apply and pending_writes:
|
|
backups: list[tuple[Path, bytes | None]] = []
|
|
try:
|
|
for destination, data, row in pending_writes:
|
|
destination.parent.mkdir(parents=True, exist_ok=True)
|
|
backups.append((destination, destination.read_bytes() if destination.exists() else None))
|
|
fd, temp_name = tempfile.mkstemp(prefix=f".{destination.name}.", dir=destination.parent)
|
|
try:
|
|
with os.fdopen(fd, "wb") as handle:
|
|
handle.write(data)
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
os.chmod(temp_name, 0o644)
|
|
os.replace(temp_name, destination)
|
|
finally:
|
|
if os.path.exists(temp_name):
|
|
os.unlink(temp_name)
|
|
row["status"] = "applied"
|
|
except Exception:
|
|
for destination, previous in reversed(backups):
|
|
if previous is None:
|
|
destination.unlink(missing_ok=True)
|
|
else:
|
|
destination.write_bytes(previous)
|
|
raise
|
|
|
|
return {
|
|
"schema_version": 1,
|
|
"source_repository": policy["source_repository"],
|
|
"source_commit": source_commit,
|
|
"source_tree_hash": source_tree_hash(source),
|
|
"source_mode": source_mode,
|
|
"target_repository": policy["target_repository"],
|
|
"target_base_commit": target_commit.lower(),
|
|
"policy_version": 2,
|
|
"generated_at": datetime.now(UTC).isoformat().replace("+00:00", "Z"),
|
|
"mirror_files": mirror_files,
|
|
"semantic_merge_files": semantic_files,
|
|
"protected_rejections": [],
|
|
"tests_run": [],
|
|
"privacy_scan": {"status": "pass", "scanned_files": len(mirror_files), "rejections": []},
|
|
"operator_review_required": any(row["status"] in {"review_required", "target_missing"} for row in semantic_files),
|
|
}
|
|
|
|
|
|
def write_json_atomic(path: Path, value: dict[str, Any]) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
data = (json.dumps(value, ensure_ascii=False, indent=2, sort_keys=True) + "\n").encode("utf-8")
|
|
fd, temp_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent)
|
|
try:
|
|
with os.fdopen(fd, "wb") as handle:
|
|
handle.write(data)
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
os.replace(temp_name, path)
|
|
finally:
|
|
if os.path.exists(temp_name):
|
|
os.unlink(temp_name)
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("--source", type=Path, required=True)
|
|
parser.add_argument("--policy", type=Path, default=DEFAULT_POLICY)
|
|
mode = parser.add_mutually_exclusive_group()
|
|
mode.add_argument("--dry-run", action="store_true")
|
|
mode.add_argument("--apply", action="store_true")
|
|
parser.add_argument("--output", type=Path, required=True)
|
|
args = parser.parse_args(argv)
|
|
try:
|
|
manifest = build_manifest(args.source, policy_path=args.policy, apply=args.apply)
|
|
write_json_atomic(args.output, manifest)
|
|
except ImportRejected as error:
|
|
print(json.dumps({"status": "rejected", "reason": str(error)}, ensure_ascii=False), file=sys.stderr)
|
|
return EXIT_POLICY_REJECTED
|
|
except (OSError, ValueError, json.JSONDecodeError) as error:
|
|
print(json.dumps({"status": "invalid", "reason": str(error)}, ensure_ascii=False), file=sys.stderr)
|
|
return EXIT_VALIDATION_FAILED
|
|
print(json.dumps(manifest, ensure_ascii=False, indent=2, sort_keys=True))
|
|
return EXIT_SEMANTIC_REVIEW if manifest["operator_review_required"] else EXIT_OK
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|