인수인계 폴더의 필수 파일과 예상 밖 파일을 계약표로 확인하기

"""Check a stable, owned Linux practice tree; write only a new JSON report."""
import argparse
import hashlib
import json
import os
from pathlib import Path, PureWindowsPath
import stat
import sys

MAX_ENTRIES = 2000
MAX_DEPTH = 16
MAX_CONTRACT_BYTES = 32768
MAX_REQUIRED = 64


def plain_path(path):
    path = Path(os.path.abspath(path))
    for part in [*reversed(path.parents), path]:
        if stat.S_ISLNK(part.lstat().st_mode):
            raise ValueError(f"symlink path component: {part}")
    return path


def validate_contract(obj):
    if not isinstance(obj, dict) or set(obj) != {"contract_version", "required"}:
        raise ValueError("contract needs only contract_version and required")
    version, rows = obj["contract_version"], obj["required"]
    if not isinstance(version, str) or not version.strip() or len(version) > 80:
        raise ValueError("contract_version must be a nonempty string <= 80 chars")
    if not isinstance(rows, list) or not 1 <= len(rows) <= MAX_REQUIRED:
        raise ValueError("required must contain 1..64 entries")
    required, prefixes = {}, {}
    for row in rows:
        if not isinstance(row, dict) or set(row) != {"path", "kind"}:
            raise ValueError("each required entry needs path and kind")
        name, kind = row["path"], row["kind"]
        if (not isinstance(name, str) or not name or len(name) > 240
                or name.startswith("/") or "\\" in name or ":" in name
                or PureWindowsPath(name).drive
                or any(ord(c) < 32 or 0xD800 <= ord(c) <= 0xDFFF for c in name)):
            raise ValueError("use a plain relative path with forward slashes")
        parts = name.split("/")
        if any(not p or p in (".", "..") for p in parts):
            raise ValueError("empty, . and .. path components are forbidden")
        if kind not in ("file", "dir") or name in required:
            raise ValueError("invalid kind or duplicate path")
        required[name] = kind
        for n in range(1, len(parts) + 1):
            prefix = "/".join(parts[:n])
            folded = prefix.casefold()
            if folded in prefixes and prefixes[folded] != prefix:
                raise ValueError("casefold collision in contract paths")
            prefixes[folded] = prefix
    for name in required:
        for parent in Path(name).parents:
            if str(parent) != "." and required.get(parent.as_posix()) == "file":
                raise ValueError("a required file cannot be another path's parent")
    return version, required


def unique_object(pairs):
    obj = {}
    for key, value in pairs:
        if key in obj:
            raise ValueError("duplicate JSON key in contract")
        obj[key] = value
    return obj


def load_contract(path, expected_sha256):
    path = plain_path(path)
    if not stat.S_ISREG(path.lstat().st_mode):
        raise ValueError("contract must be a regular file")
    if len(expected_sha256) != 64 or any(c not in "0123456789abcdef" for c in expected_sha256):
        raise ValueError("expected SHA-256 must be 64 lowercase hex chars")
    with path.open("rb") as f:
        raw = f.read(MAX_CONTRACT_BYTES + 1)
    if len(raw) > MAX_CONTRACT_BYTES:
        raise ValueError("contract exceeds 32768 bytes")
    actual = hashlib.sha256(raw).hexdigest()
    if actual != expected_sha256:
        raise ValueError("contract SHA-256 differs; confirm the approved contract")
    version, required = validate_contract(json.loads(raw.decode("utf-8"), object_pairs_hook=unique_object))
    return version, required, actual


def scan_tree(root):
    observed, issues = {}, []
    folded_names = {}
    attempted_entries = 0

    def issue(name, operation, detail):
        issues.append({"path": name, "operation": operation, "detail": detail})

    def visit(fd, prefix, depth):
        nonlocal attempted_entries
        if depth >= MAX_DEPTH:
            issue(prefix, "depth_limit", "children beyond maximum entry depth 16 not observed")
            return
        try:
            with os.scandir(fd) as entries:
                for entry in entries:
                    if attempted_entries >= MAX_ENTRIES:
                        issue(prefix, "entry_limit", "maximum attempted entries 2000 reached")
                        return
                    attempted_entries += 1
                    name = f"{prefix}/{entry.name}" if prefix else entry.name
                    try:
                        info = os.stat(entry.name, dir_fd=fd, follow_symlinks=False)
                    except OSError as exc:
                        issue(name, "stat", str(exc))
                        continue
                    if any(0xD800 <= ord(c) <= 0xDFFF for c in name):
                        issue(name, "non_utf8_name", "filename is not valid UTF-8")
                    mode = info.st_mode
                    kind = ("file" if stat.S_ISREG(mode) else "dir" if stat.S_ISDIR(mode)
                            else "symlink" if stat.S_ISLNK(mode) else "special")
                    observed[name] = kind
                    key = name.casefold()
                    if key in folded_names and folded_names[key] != name:
                        issue(name, "case_collision", f"also observed: {folded_names[key]}")
                    folded_names[key] = name
                    if kind not in ("file", "dir"):
                        issue(name, "unsupported_kind", kind)
                        continue
                    flags = os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK
                    if kind == "dir":
                        flags |= os.O_DIRECTORY
                    child = None
                    try:
                        child = os.open(entry.name, flags, dir_fd=fd)
                        opened = os.fstat(child)
                        if (opened.st_dev, opened.st_ino, stat.S_IFMT(opened.st_mode)) != (
                                info.st_dev, info.st_ino, stat.S_IFMT(info.st_mode)):
                            issue(name, "changed_entry", "entry changed while opening")
                        elif kind == "dir":
                            visit(child, name, depth + 1)
                        else:
                            os.read(child, 1)  # Readability probe only; no format/content validation.
                    except OSError as exc:
                        issue(name, "scandir" if kind == "dir" else "read_probe", str(exc))
                    finally:
                        if child is not None:
                            os.close(child)
        except OSError as exc:
            issue(prefix, "scandir", str(exc))

    fd = None
    try:
        fd = os.open(root, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW)
        visit(fd, "", 0)
    except OSError as exc:
        issue("", "open_root", str(exc))
    finally:
        if fd is not None:
            os.close(fd)
    return observed, issues


def compare(root, version, required, contract_sha256):
    expected = dict(required)
    for name in required:
        for parent in Path(name).parents:
            if str(parent) != ".":
                expected.setdefault(parent.as_posix(), "dir")
    observed, issues = scan_tree(root)
    def uncertain(name):
        return any(not x["path"] or name == x["path"] or name.startswith(x["path"] + "/") for x in issues)
    missing = sorted(p for p in required if p not in observed and not uncertain(p))
    unverified = sorted(p for p in required if uncertain(p))
    unexpected = [{"path": p, "kind": observed[p]} for p in sorted(observed) if p not in expected]
    mismatch = [{"path": p, "expected": expected[p], "observed": observed[p], "explicit": p in required}
                for p in sorted(expected.keys() & observed.keys()) if expected[p] != observed[p]]
    status = "INCOMPLETE" if issues else "REVIEW" if missing or unexpected or mismatch else "PASS"
    return {"status": status, "contract_version": version, "contract_sha256": contract_sha256,
            "required_count": len(required), "observed_count": len(observed),
            "scope": {"hidden_entries": "included", "symlinks": "recorded_not_followed",
                      "maximum_depth": MAX_DEPTH, "maximum_entries": MAX_ENTRIES,
                      "content_validation": False, "readability_probe_bytes": 1,
                      "atomic_snapshot": False},
            "missing": missing, "unexpected": unexpected, "type_mismatches": mismatch,
            "unverified_required": unverified,
            "scan_issues": sorted(issues, key=lambda x: (x["path"], x["operation"])),
            "observed": [{"path": p, "kind": observed[p]} for p in sorted(observed)]}


def validate_output(root, output):
    root = plain_path(root)
    output = Path(os.path.abspath(output))
    parent = plain_path(output.parent)
    if not parent.is_dir():
        raise ValueError("output parent must already exist")
    if output == root or root in output.parents:
        raise ValueError("output must be outside the observed root")
    try:
        output.lstat()
    except FileNotFoundError:
        return output
    raise ValueError("output already exists; choose a new report name")


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("root", type=Path)
    parser.add_argument("contract", type=Path)
    parser.add_argument("output", type=Path)
    parser.add_argument("--expected-sha256", required=True)
    args = parser.parse_args()
    try:
        if not sys.platform.startswith("linux"):
            raise ValueError("this practice checker is Linux-only; other OSes were not validated")
        root = plain_path(args.root)
        if not stat.S_ISDIR(root.lstat().st_mode):
            raise ValueError("root must be a regular directory")
        output = validate_output(root, args.output)
        contract = plain_path(args.contract)
        if contract == root or root in contract.parents:
            raise ValueError("contract must be outside the observed root")
        version, required, sha = load_contract(contract, args.expected_sha256)
        report = compare(root, version, required, sha)
        payload = json.dumps(report, ensure_ascii=True, indent=2) + "\n"
        with output.open("x", encoding="utf-8", newline="\n") as f:
            f.write(payload)
        print(f'{report["status"]}: {output.name}')
        return {"PASS": 0, "REVIEW": 1, "INCOMPLETE": 3}[report["status"]]
    except (OSError, ValueError, UnicodeError) as exc:
        print(f"STOP: {exc}", file=sys.stderr)
        return 2


if __name__ == "__main__":
    raise SystemExit(main())