#!/usr/bin/env python3
from __future__ import annotations

import argparse
import hashlib
import json
from pathlib import Path
import re
import subprocess
import tempfile


VTYPE_NUMBER = {
    "VNON": 0,
    "VREG": 1,
    "VDIR": 2,
    "VBLK": 3,
    "VCHR": 4,
    "VLNK": 5,
    "VSOCK": 6,
    "VFIFO": 7,
}


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser()
    parser.add_argument("--root", type=Path, required=True)
    parser.add_argument("--dut", type=Path, required=True)
    parser.add_argument("--spec", type=Path, required=True)
    parser.add_argument("--gate-output", type=Path, required=True)
    parser.add_argument("--artifacts", type=Path, required=True)
    return parser.parse_args()


def sha256(path: Path) -> str:
    return hashlib.sha256(path.read_bytes()).hexdigest()


def committed(root: Path, commit: str, path: str) -> str:
    completed = subprocess.run(
        ["git", "-C", str(root), "show", f"{commit}:{path}"],
        check=False,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )
    if completed.returncode != 0:
        raise SystemExit(f"cannot read {path} at {commit}: {completed.stderr}")
    return completed.stdout


def replace_once(source: str, old: str, new: str, label: str) -> str:
    count = source.count(old)
    if count != 1:
        raise SystemExit(f"{label}: expected one transform site, found {count}")
    return source.replace(old, new, 1)


def extract_function(source: str, name: str) -> str:
    match = re.search(r"^" + re.escape(name) + r"\s*\(", source, re.MULTILINE)
    if match is None:
        raise SystemExit(f"function not found: {name}")
    start = source.rfind("\n\n", 0, match.start()) + 2
    brace = source.find("{", match.end())
    if brace < 0:
        raise SystemExit(f"function body not found: {name}")
    depth = 0
    state = "code"
    index = brace
    while index < len(source):
        char = source[index]
        following = source[index + 1] if index + 1 < len(source) else ""
        if state == "code":
            if char == "/" and following == "*":
                state = "block"
                index += 2
                continue
            if char == "/" and following == "/":
                state = "line"
                index += 2
                continue
            if char == '"':
                state = "string"
            elif char == "'":
                state = "character"
            elif char == "{":
                depth += 1
            elif char == "}":
                depth -= 1
                if depth == 0:
                    return source[start : index + 1] + "\n"
        elif state == "block" and char == "*" and following == "/":
            state = "code"
            index += 2
            continue
        elif state == "line" and char == "\n":
            state = "code"
        elif state in {"string", "character"}:
            if char == "\\":
                index += 2
                continue
            if (state == "string" and char == '"') or (
                state == "character" and char == "'"
            ):
                state = "code"
        index += 1
    raise SystemExit(f"unterminated function: {name}")


def run_helper(helper: str, records: list[dict], artifacts: Path) -> list[dict]:
    harness = f'''#include <stdbool.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>

#define EROFS_FT_UNKNOWN 0
#define EROFS_FT_REG_FILE 1
#define EROFS_FT_DIR 2
#define EROFS_FT_CHRDEV 3
#define EROFS_FT_BLKDEV 4
#define EROFS_FT_FIFO 5
#define EROFS_FT_SOCK 6
#define EROFS_FT_SYMLINK 7
#define __enum_uint8(name) enum name

enum vtype {{
    VNON,
    VREG,
    VDIR,
    VBLK,
    VCHR,
    VLNK,
    VSOCK,
    VFIFO,
}};

{helper}
int
main(int argc, char **argv)
{{
    unsigned long file_type, vtype;

    if (argc != 3)
        return (2);
    file_type = strtoul(argv[1], NULL, 0);
    vtype = strtoul(argv[2], NULL, 0);
    if (file_type > UINT8_MAX || vtype > VFIFO)
        return (2);
    printf("%d\\n", erofs_dirent_type_matches((uint8_t)file_type,
        (enum vtype)vtype) ? 0 : 97);
    return (0);
}}
'''
    source = artifacts / "B11-helper-harness.c"
    source.write_text(harness, encoding="ascii")
    with tempfile.TemporaryDirectory(prefix="b11-helper-") as temporary:
        binary = Path(temporary) / "helper"
        completed = subprocess.run(
            [
                "cc",
                "-std=c11",
                "-Wall",
                "-Wextra",
                "-Werror",
                str(source),
                "-o",
                str(binary),
            ],
            check=False,
            text=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
        )
        if completed.returncode != 0:
            raise SystemExit(f"B11 helper harness did not compile:\n{completed.stdout}")
        results = []
        for record in records:
            completed = subprocess.run(
                [
                    str(binary),
                    str(record["file_type"]),
                    str(VTYPE_NUMBER[record["vtype"]]),
                ],
                check=False,
                text=True,
                stdout=subprocess.PIPE,
                stderr=subprocess.STDOUT,
            )
            if completed.returncode != 0:
                raise SystemExit(f"B11 helper failed for {record['id']}")
            actual = int(completed.stdout.strip())
            expected = record["expected_errno"]
            if actual != expected:
                raise SystemExit(
                    f"B11 helper result differs for {record['id']}: {actual} != {expected}"
                )
            results.append(
                {
                    "actual_errno": actual,
                    "expected_errno": expected,
                    "id": record["id"],
                }
            )
    return results


def main() -> None:
    args = parse_args()
    spec = json.loads(args.spec.read_text(encoding="ascii"))
    if spec.get("schema") != 1 or spec.get("batch") != "B11":
        raise SystemExit("invalid B11 dtype spec")
    args.artifacts.mkdir(parents=True, exist_ok=True)
    gate_decision = subprocess.check_output(
        ["git", "-C", str(args.root), "rev-parse", spec["gate_decision"]],
        text=True,
    ).strip()
    if gate_decision != spec["gate_decision"]:
        raise SystemExit("B11 gate decision identity changed")

    gate_script = args.dut / "tests/pre15/gates/P15-081.sh"
    gate_input = args.dut / "tests/pre15/gates/P15-081-input.json"
    if sha256(gate_script) != spec["gate_script_sha256"]:
        raise SystemExit("B11 gate script identity changed")
    if sha256(gate_input) != spec["gate_input_sha256"]:
        raise SystemExit("B11 gate input identity changed")
    for name, expected in spec["gate_evidence_sha256"].items():
        path = args.gate_output / name
        if sha256(path) != expected:
            raise SystemExit(f"B11 gate replay identity changed: {name}")

    gate_result = json.loads(
        (args.gate_output / "result.json").read_text(encoding="ascii")
    )
    gate_oracle = json.loads(
        (args.gate_output / "oracle.json").read_text(encoding="ascii")
    )
    if gate_result["status"] != "GO" or gate_result["b11"] != "AUTHORIZED":
        raise SystemExit("P15-081 gate no longer authorizes B11")
    for key, expected in spec["expected"].items():
        actual = (
            gate_result["fixture_set_sha256"]
            if key == "fixture_set_sha256"
            else gate_result[key]
        )
        if actual != expected:
            raise SystemExit(f"B11 gate count changed for {key}: {actual} != {expected}")

    baseline = spec["baseline"]
    paths = ("inode.c", "internal.h", "namei.c", "dir.c", "erofs_vnops.c")
    before = {
        name: committed(args.root, baseline, f"repo-pre-15/src/{name}")
        for name in paths
    }
    current = {
        name: (args.dut / "src" / name).read_text(encoding="utf-8")
        for name in paths
    }

    helper = '''bool
erofs_dirent_type_matches(uint8_t file_type, __enum_uint8(vtype) vtype)
{
\tswitch (file_type) {
\tcase EROFS_FT_REG_FILE:
\t\treturn (vtype == VREG);
\tcase EROFS_FT_DIR:
\t\treturn (vtype == VDIR);
\tcase EROFS_FT_CHRDEV:
\t\treturn (vtype == VCHR);
\tcase EROFS_FT_BLKDEV:
\t\treturn (vtype == VBLK);
\tcase EROFS_FT_FIFO:
\t\treturn (vtype == VFIFO);
\tcase EROFS_FT_SOCK:
\t\treturn (vtype == VSOCK);
\tcase EROFS_FT_SYMLINK:
\t\treturn (vtype == VLNK);
\tdefault:
\t\treturn (true);
\t}
}

'''
    inode_marker = "/*\n * Read and decode a disk inode."
    expected_inode = replace_once(
        before["inode.c"], inode_marker, helper + inode_marker, "inode helper"
    )
    expected_internal = replace_once(
        before["internal.h"],
        "int erofs_read_inode(struct erofs_sb_info *sbi, erofs_nid_t nid,\n",
        "bool erofs_dirent_type_matches(uint8_t file_type,\n"
        "    __enum_uint8(vtype) vtype);\n"
        "int erofs_read_inode(struct erofs_sb_info *sbi, erofs_nid_t nid,\n",
        "internal prototype",
    )
    lookup_marker = "\tif (error != 0)\n\t\treturn (error);\n\t*ap->a_vpp = vp;\n"
    lookup_replacement = (
        "\tif (error != 0)\n"
        "\t\treturn (error);\n"
        "\tif ((cnp->cn_flags & ISDOTDOT) == 0 &&\n"
        "\t    !erofs_dirent_type_matches(dtype, VTOE(vp)->vtype)) {\n"
        "\t\tvput(vp);\n"
        "\t\treturn (EINTEGRITY);\n"
        "\t}\n"
        "\t*ap->a_vpp = vp;\n"
    )
    expected_namei = replace_once(
        before["namei.c"], lookup_marker, lookup_replacement, "lookup validator"
    )
    expected = {
        "inode.c": expected_inode,
        "internal.h": expected_internal,
        "namei.c": expected_namei,
        "dir.c": before["dir.c"],
        "erofs_vnops.c": before["erofs_vnops.c"],
    }
    for name in paths:
        if current[name] != expected[name]:
            raise SystemExit(f"B11 source differs from the exact transform: {name}")

    lookup = extract_function(current["namei.c"], "erofs_lookup")
    validator = lookup.index("erofs_dirent_type_matches")
    if lookup.index("error = erofs_vget(") > validator:
        raise SystemExit("B11 validator moved before the ordinary vnode lookup")
    if lookup.index("error = vn_vget_ino(") > validator:
        raise SystemExit("B11 validator moved before the dotdot vnode lookup")
    if validator > lookup.index("*ap->a_vpp = vp"):
        raise SystemExit("B11 validator moved after vnode publication")
    if validator > lookup.index("cache_enter(dvp, vp, cnp)"):
        raise SystemExit("B11 validator moved after namecache publication")
    validator_start = lookup.rfind(
        "if ((cnp->cn_flags & ISDOTDOT) == 0", 0, validator
    )
    if validator_start < 0:
        raise SystemExit("B11 validator condition is absent")
    validator_block = lookup[validator_start : lookup.index("*ap->a_vpp = vp")]
    if "(cnp->cn_flags & ISDOTDOT) == 0" not in validator_block:
        raise SystemExit("B11 dotdot bypass is absent")
    if "vput(vp);\n\t\treturn (EINTEGRITY);" not in validator_block:
        raise SystemExit("B11 mismatch cleanup or positive errno changed")
    for token in ("erofs_vget(", "vn_vget_ino(", "vfs_hash_get(", "VOP_LOCK(", "vn_lock("):
        if token in validator_block:
            raise SystemExit(f"B11 validator introduced a lock/vget operation: {token}")
    for token in ("erofs_vget(", "vn_vget_ino(", "vfs_hash_get("):
        if token in current["dir.c"]:
            raise SystemExit(f"B11 readdir path introduced vnode loading: {token}")

    actual_helper = extract_function(current["inode.c"], "erofs_dirent_type_matches")
    records = []
    for record in gate_oracle["known_match_records"]:
        records.append(
            {
                "expected_errno": record["expected_errno"],
                "file_type": record["file_type"],
                "id": record["case"],
                "vtype": record["vtype"],
            }
        )
    for record in gate_oracle["case_records"]:
        records.append(
            {
                "expected_errno": record["expected_errno"],
                "file_type": record["file_type"],
                "id": record["id"],
                "vtype": record["vtype"],
            }
        )
    helper_results = run_helper(actual_helper, records, args.artifacts)

    report = {
        "batch": "B11",
        "candidate": "P15-081",
        "dir_source_unchanged": True,
        "dotdot_bypasses_validator": True,
        "gate_status": gate_result["status"],
        "helper_case_count": len(helper_results),
        "helper_results": helper_results,
        "known_mismatch_errno": "+EINTEGRITY",
        "namecache_publication_after_validator": True,
        "readdir_vget_count": 0,
        "source_transform_exact": True,
        "status": "PASS",
        "unknown_and_reserved_tolerated": True,
        "vnode_source_unchanged": True,
    }
    (args.artifacts / "B11-dtype-report.json").write_text(
        json.dumps(report, indent=2, sort_keys=True) + "\n", encoding="ascii"
    )
    print(
        "B11 dtype: PASS "
        f"helper_cases={len(helper_results)} known_mismatch=+EINTEGRITY "
        "unknown=accepted readdir_vgets=0 dotdot=bypass"
    )


if __name__ == "__main__":
    main()
