Jyotisha/scripts/import_yinduzhanxing.py

#!/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. Non-Git archives are accepted
only when an operator binds them to an explicit commit and expected source-tree
SHA-256; missing or mismatched provenance fails closed.
"""

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
from urllib.parse import urlsplit

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
COMMIT_HEX_LENGTH = 40
SHA256_HEX_LENGTH = 64
IMPORT_RECORD_DIR = ROOT / "references/cross_project_contract/imports"
REQUIRED_PROTECTED_PATTERNS = frozenset({
    "skills/jyotish-birth-time-rectification/**",
    "skills/jyotish-vedic-astrology/**",
    "SKILL.md",
    "frontend/**",
    "deploy/**",
    ".gitea/**",
    ".github/**",
    "frontend/db/**",
    "frontend/supabase/**",
    "references/oracle/commercial_skill_truth_overlay.v1.json",
    "**/.env",
    "**/.env.*",
    "**/*payment*",
    "**/*billing*",
    "**/*subscription*",
    "**/*entitlement*",
    "**/*admin*",
    "**/*service_role*",
})


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:
    root_resolved = root.resolve()
    safe_relative = _safe_relative(relative)
    path = root_resolved / safe_relative
    current = root_resolved
    for part in PurePosixPath(safe_relative).parts:
        current = current / part
        if current.is_symlink():
            raise ImportRejected(f"symlink_component_rejected:{relative}")
    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")
    expected_identity = _repository_identity(str(policy.get("source_repository_url", "")))
    if expected_identity is None:
        raise ImportRejected("source_repository_url_invalid")
    if expected_identity[1] != str(policy.get("source_repository", "")).strip("/").casefold():
        raise ImportRejected("source_repository_url_slug_mismatch")
    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"]]
    missing_protected = sorted(REQUIRED_PROTECTED_PATTERNS - set(protected))
    if missing_protected:
        raise ImportRejected(f"required_protected_patterns_missing:{','.join(missing_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 _normalized_hex(value: str | None, *, length: int, label: str) -> str | None:
    if value is None:
        return None
    normalized = value.strip().lower()
    if len(normalized) != length or any(ch not in "0123456789abcdef" for ch in normalized):
        raise ImportRejected(f"{label}_invalid")
    return normalized


def source_identity(
    source: Path,
    *,
    source_mode: str = "auto",
    source_commit: str | None = None,
    expected_tree_sha256: str | None = None,
    actual_tree_sha256: str | None = None,
) -> tuple[str, str]:
    if source_mode not in {"auto", "git", "archive"}:
        raise ImportRejected("source_mode_must_be_auto_git_or_archive")
    if source_mode != "archive" and (source_commit is None) != (expected_tree_sha256 is None):
        raise ImportRejected("source_commit_and_expected_tree_sha256_must_be_provided_together")

    explicit_commit = _normalized_hex(
        source_commit, length=COMMIT_HEX_LENGTH, label="source_commit"
    )
    expected_tree = _normalized_hex(
        expected_tree_sha256, length=SHA256_HEX_LENGTH, label="expected_tree_sha256"
    )
    actual_tree = actual_tree_sha256 or source_tree_hash(source)

    inside = _git_output(source, "rev-parse", "--show-toplevel")
    detected_commit = (
        _git_output(source, "rev-parse", "HEAD")
        if inside and Path(inside).resolve() == source.resolve()
        else None
    )
    detected_commit = _normalized_hex(
        detected_commit, length=COMMIT_HEX_LENGTH, label="detected_source_commit"
    )

    if source_mode == "archive" and detected_commit:
        raise ImportRejected("archive_source_must_not_be_git_checkout")

    if source_mode != "archive" and detected_commit:
        status = _git_output(source, "status", "--porcelain=v1", "--untracked-files=all")
        ignored = _git_output(source, "ls-files", "--others", "--ignored", "--exclude-standard")
        if status is None or ignored is None:
            raise ImportRejected("git_source_status_unavailable")
        if status or ignored:
            raise ImportRejected("git_source_must_be_clean")
        if explicit_commit and explicit_commit != detected_commit:
            raise ImportRejected("source_commit_mismatch")
        if expected_tree and expected_tree != actual_tree:
            raise ImportRejected("source_tree_sha256_mismatch")
        return "git", detected_commit

    if source_mode == "git":
        raise ImportRejected("git_source_metadata_unavailable")
    if explicit_commit is None:
        raise ImportRejected("archive_source_commit_required")
    if expected_tree is None:
        raise ImportRejected("archive_expected_tree_sha256_required")
    if expected_tree != actual_tree:
        raise ImportRejected("source_tree_sha256_mismatch")
    return "archive", explicit_commit


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 _repository_identity(value: str) -> tuple[str, str] | None:
    value = value.strip().removesuffix(".git").rstrip("/")
    if not value:
        return None
    if "://" in value:
        parsed = urlsplit(value)
        host = (parsed.hostname or "").casefold()
        path = parsed.path
    elif ":" in value and "@" in value.split(":", 1)[0]:
        host = value.split("@", 1)[1].split(":", 1)[0].casefold()
        path = value.split(":", 1)[1]
    else:
        return None
    parts = [part for part in path.split("/") if part]
    if not host or len(parts) < 2:
        return None
    return host, "/".join(parts).casefold()


def _remote_repository_identity(root: Path) -> tuple[str, str] | None:
    remote = _git_output(root, "remote", "get-url", "origin")
    return _repository_identity(remote or "")


def validate_git_source_repository(source: Path, expected_repository: str, expected_url: str) -> None:
    expected = expected_repository.strip().removesuffix(".git").strip("/").casefold()
    expected_identity = _repository_identity(expected_url)
    actual = _remote_repository_identity(source)
    if actual is None:
        raise ImportRejected("git_source_origin_unavailable")
    if expected_identity is None or expected_identity[1] != expected:
        raise ImportRejected("source_repository_url_invalid")
    if actual != expected_identity:
        raise ImportRejected(f"git_source_repository_mismatch:{actual[0]}/{actual[1]}")


def validate_clean_target(target: Path) -> None:
    top_level = _git_output(target, "rev-parse", "--show-toplevel")
    if not top_level or Path(top_level).resolve() != target.resolve():
        raise ImportRejected("target_must_be_git_repository_root")
    status = _git_output(target, "status", "--porcelain=v1", "--untracked-files=all")
    ignored = _git_output(target, "ls-files", "--others", "--ignored", "--exclude-standard")
    if status is None or ignored is None:
        raise ImportRejected("target_status_unavailable")
    if status or ignored:
        raise ImportRejected("target_must_be_clean")


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:
            continue
        if path.is_symlink():
            raise ImportRejected(f"source_tree_symlink_rejected:{relative}")
        if 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 _normalized_commit_timestamp(value: str | None) -> str | None:
    if not value:
        return None
    try:
        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
    except ValueError:
        return None
    if parsed.tzinfo is None:
        return None
    return parsed.astimezone(UTC).isoformat(timespec="seconds").replace("+00:00", "Z")


def deterministic_generated_at(
    source: Path,
    *,
    source_mode: str,
    source_commit: str,
    target: Path,
    target_commit: str,
) -> str:
    timestamp = None
    if source_mode == "git":
        timestamp = _git_output(source, "show", "-s", "--format=%cI", source_commit)
    if timestamp is None:
        timestamp = _git_output(target, "show", "-s", "--format=%cI", target_commit)
    normalized = _normalized_commit_timestamp(timestamp)
    if normalized is None:
        raise ImportRejected("deterministic_generated_at_unavailable")
    return normalized


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,
    source_mode: str = "auto",
    source_commit: str | None = None,
    expected_tree_sha256: str | None = None,
    tests_run: list[str] | None = None,
) -> dict[str, Any]:
    source = source.expanduser().resolve(strict=True)
    target = target.expanduser().resolve(strict=True)
    policy = load_policy(policy_path)
    validate_roots(source, target)
    validate_clean_target(target)
    protected = list(policy["modes"]["protected"])
    privacy = policy.get("privacy") if isinstance(policy.get("privacy"), dict) else {}
    actual_tree_sha256 = source_tree_hash(source)
    resolved_source_mode, resolved_source_commit = source_identity(
        source,
        source_mode=source_mode,
        source_commit=source_commit,
        expected_tree_sha256=expected_tree_sha256,
        actual_tree_sha256=actual_tree_sha256,
    )
    if resolved_source_mode == "git":
        validate_git_source_repository(
            source,
            str(policy["source_repository"]),
            str(policy["source_repository_url"]),
        )
    source_skill_file = _resolved_file(source, "SKILL.md", must_exist=True)
    if not source_skill_file.exists():
        raise ImportRejected("root_skill_missing")
    source_skill_sha256 = sha256_file(source_skill_file)
    target_commit = _normalized_hex(
        _git_output(target, "rev-parse", "HEAD"),
        length=COMMIT_HEX_LENGTH,
        label="target_base_commit",
    )
    if target_commit is None:
        raise ImportRejected("target_base_commit_unavailable")
    generated_at = deterministic_generated_at(
        source,
        source_mode=resolved_source_mode,
        source_commit=resolved_source_commit,
        target=target,
        target_commit=target_commit,
    )

    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_repository_url": policy["source_repository_url"],
        "source_commit": resolved_source_commit,
        "source_tree_hash": actual_tree_sha256,
        "source_skill_sha256": source_skill_sha256,
        "source_mode": resolved_source_mode,
        "target_repository": policy["target_repository"],
        "target_base_commit": target_commit,
        "policy_version": 2,
        "generated_at": generated_at,
        "mirror_files": mirror_files,
        "semantic_merge_files": semantic_files,
        "protected_rejections": [],
        "tests_run": sorted(set(tests_run or [])),
        "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 validate_output_path(path: Path, *, target: Path = ROOT, source: Path | None = None) -> Path:
    """Allow manifests outside the target or inside the dedicated import-record directory only."""
    resolved = path.expanduser().resolve(strict=False)
    target_resolved = target.expanduser().resolve(strict=True)
    if source is not None:
        source_resolved = source.expanduser().resolve(strict=True)
        if resolved == source_resolved or source_resolved in resolved.parents:
            raise ImportRejected("manifest_output_must_not_overlap_source")
    if resolved == target_resolved or target_resolved in resolved.parents:
        import_root = (target_resolved / IMPORT_RECORD_DIR.relative_to(ROOT)).resolve(strict=False)
        if import_root not in resolved.parents:
            raise ImportRejected("manifest_output_must_be_external_or_import_record_json")
    if resolved.suffix.casefold() != ".json":
        raise ImportRejected("manifest_output_must_be_json")
    return resolved


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)
    parser.add_argument("--source-mode", choices=("auto", "git", "archive"), default="auto")
    parser.add_argument("--source-commit")
    parser.add_argument("--expected-tree-sha256")
    parser.add_argument("--test-run", action="append", default=[], help="Test path recorded in the manifest; repeatable.")
    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:
        output = validate_output_path(args.output, source=args.source)
        manifest = build_manifest(
            args.source,
            policy_path=args.policy,
            apply=args.apply,
            source_mode=args.source_mode,
            source_commit=args.source_commit,
            expected_tree_sha256=args.expected_tree_sha256,
            tests_run=args.test_run,
        )
        write_json_atomic(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())