# -*- coding: utf-8 -*-
"""dispatch.anu_executor_provenance — task-2753 Option B provenance-path validator.

Layer A / pure validation, side-effect 0.

envelope(ANU artifact) **부재** 인 untrusted executor_result 를 ANU 가 수거하기 전
**provenance hard-fail 검증** 만 수행하는 순수 함수 모듈. side-effect 0:
  - 파일 write 0 (read-only: capability snapshot + result 파일 sha256 read 만).
  - cron register/remove 0, subprocess 0, merge 0, git/gh import 0.
  - owner_proof_pickup_gate(callback fire 보호) 미참조 — 이 모듈은 collector_result
    생성 여부(=수거 진행/quarantine/skip/duplicate) 만 판정한다. fire 는 frozen gate 경유.

★ raw key 0: 실제 ANU/dev key literal 을 **절대 하드코딩하지 않는다**.
  forbidden key literal 은 ``forbidden_key_literals`` 파라미터로 주입받는다.

신뢰 앵커 = ``memory/capabilities/<task_id>.json`` (dispatch 가 생성·봇 변조 불가:
forbidden_paths 자동 등록). 단일 신호 금지 — 14조건 전부 hard-fail(=quarantine).
"""
from __future__ import annotations

import hashlib
import json
import os
import re
from dataclasses import dataclass
from typing import Optional

# ── provenance reason enum (design packet §3 과 정확히 일치) ───────────────────
REASON_NO_DISPATCH_RECORD = "provenance_no_dispatch_record"   # 조건1
REASON_TASK_ID_MISMATCH = "provenance_task_id_mismatch"       # 조건2
REASON_CAPSULE_SHA_MISMATCH = "provenance_capsule_sha_mismatch"  # 조건3
REASON_SOURCE_PATH_INVALID = "provenance_source_path_invalid"   # 조건4
REASON_SCHEMA_FAIL = "schema_fail"                           # 조건5
REASON_TERMINAL_INVALID = "provenance_terminal_invalid"      # 조건6
REASON_COMPLETION_INVALID = "provenance_completion_invalid"  # 조건7
REASON_PRE_ACTIVATION_EPOCH = "pre_activation_epoch"         # 조건9 (skip)
REASON_DUPLICATE = "duplicate"                              # 조건10 (idempotent)
REASON_CALLBACK_CLAIM = "provenance_callback_claim"         # 조건11
REASON_SELF_COLLECTOR_CLAIM = "provenance_self_collector_claim"  # 조건12
REASON_SELF_KEY_CLAIM = "provenance_self_key_claim"         # 조건13
REASON_RAW_KEY = "provenance_raw_key"                       # 조건14
REASON_TASK_ID_PATH_TRAVERSAL = "provenance_task_id_path_traversal"  # 조건0 (보안: task_id 경로 traversal)

# terminal_status 허용 값 (조건6)
VALID_TERMINAL_STATUS = ("PASS", "PASS_WITH_WARNINGS", "PARTIAL")
# completion_signal 필수 값 (조건7)
EXECUTOR_COMPLETION_SIGNAL = "EXECUTOR_RESULT_WRITTEN"

# inbox 정규 경로 구성요소 (조건4)
INBOX_REL = os.path.join("memory", "events", "p0b_inbox")


# ── ProvenanceResult ──────────────────────────────────────────────────────────
@dataclass
class ProvenanceResult:
    """provenance 검증 결과.

    - ok=True → provenance 통과 → ANU 가 collector_result(adopted_via="provenance") 생성.
    - ok=False & skip=True → stale epoch → skip(quarantine 아님, move 0/wake 0).
    - ok=False & duplicate=True → ledger 중복 → idempotent no-op.
    - ok=False (그 외) → hard-fail = quarantine (reason enum).
    """

    ok: bool                          # True면 provenance 통과 → 수거 진행
    reason: str = ""                  # ok=False일 때 quarantine reason enum
    source_result_sha256: str = ""    # 조건8: result 파일 sha256
    skip: bool = False                # 조건9: stale epoch → skip(quarantine 아님)
    duplicate: bool = False           # 조건10: ledger 중복 → idempotent no-op


# task_id 허용 문자 whitelist (기존 task id 형식: task-2753, task-2753+1 등 수용)
_TASK_ID_SAFE_RE = re.compile(r"^[A-Za-z0-9._+-]+$")


def _validate_task_id_safe(task_id: str) -> bool:
    """task_id 가 capability path 구성에 안전한지 검증.

    거부: 빈 문자열 / path separator(`/`,`\\`,os.sep/altsep) / `..` traversal sequence /
          absolute path / whitelist(`[A-Za-z0-9._+-]`) 밖 문자.
    """
    if not (isinstance(task_id, str) and task_id):
        return False
    if "/" in task_id or "\\" in task_id:
        return False
    if os.sep in task_id or (os.altsep and os.altsep in task_id):
        return False
    if ".." in task_id:
        return False
    if os.path.isabs(task_id):
        return False
    if not _TASK_ID_SAFE_RE.match(task_id):
        return False
    return True


def _safe_capability_path(task_id: str, root: str) -> Optional[str]:
    """capability path 를 안전하게 resolve. capabilities root 밖이면 None(hard-fail).

    resolve 후 realpath 가 `<root>/memory/capabilities` 하위가 아니면 None 반환
    (whitelist 통과 후에도 symlink 등 잔여 escape 를 막는 defense-in-depth).
    """
    caps_root = os.path.realpath(os.path.join(root, "memory", "capabilities"))
    candidate = os.path.realpath(os.path.join(caps_root, f"{task_id}.json"))
    try:
        if os.path.commonpath([caps_root, candidate]) != caps_root:
            return None
    except ValueError:
        # 서로 다른 drive 등 commonpath 불가 → 안전하지 않음.
        return None
    return candidate


# ── 기본 capability reader (memory/capabilities/<task_id>.json) ────────────────
def _default_capability_reader(task_id: str, root: str) -> Optional[dict]:
    """``memory/capabilities/<task_id>.json`` 을 읽어 dict 반환. 미존재/unsafe → None.

    dispatch 가 생성하는 신뢰 앵커(봇 변조 불가). read-only — side-effect 0.
    task_id traversal 방어: _safe_capability_path 로 resolve(밖이면 None).
    """
    if not _validate_task_id_safe(task_id):
        return None
    path = _safe_capability_path(task_id, root)
    if path is None or not os.path.exists(path):
        return None
    try:
        with open(path, encoding="utf-8") as f:
            return json.load(f)
    except (OSError, ValueError):
        return None


# chunked streaming hash chunk 크기 (1 MiB) — 대용량 파일 OOM 방지.
_HASH_CHUNK_SIZE = 1024 * 1024


# ── 기본 hash_fn (파일 sha256) ─────────────────────────────────────────────────
def _default_hash_fn(path: str) -> str:
    """``path`` 파일을 chunked streaming 으로 sha256 hexdigest 반환.

    전체 read 가 아니라 1 MiB chunk 로 순차 update — 대용량 파일 OOM 방지.
    결과는 전체 read sha256 과 **동일**(bit-identical). read-only — side-effect 0.
    """
    h = hashlib.sha256()
    with open(path, "rb") as f:
        for chunk in iter(lambda: f.read(_HASH_CHUNK_SIZE), b""):
            h.update(chunk)
    return h.hexdigest()


def _normalize(path: str) -> str:
    """경로 정규화 — realpath 시도(symlink 해소), 실패 시 normpath fallback."""
    try:
        return os.path.realpath(path)
    except OSError:
        return os.path.normpath(path)


def report_path_self_ref(report_path, result_path, *, root: Optional[str] = None) -> bool:
    """report_path 가 result.json 자기 자신(self-ref)을 가리키는지.

    task-2755+4 abspath 완전 정렬: result_path · report_path · anchor 를 모두
    동일한 absolute-normalized 기준(``os.path.abspath`` + ``os.path.normpath``,
    **symlink 미해소**)으로 후보 집합화한 뒤 교집합 여부로 self-ref 를 판정한다.
    상대경로는 root(제공 시)·cwd·result 디렉토리 anchor 각각으로 절대화하여,
    절대/상대 어떤 조합(abs+abs, rel+rel, abs+rel, rel+abs)이든 동일 파일이면
    self-ref 로 인식된다. realpath/symlink 는 해소하지 않는다(abspath 기준 통일).

    불변 조건:
      - report_path/result_path 가 None/빈 str/공백 → False (조건5 REASON_SCHEMA_FAIL 유지).
      - 자기 자신이 아닌 다른 report_path → False.
      - path traversal / root 밖 혼동 → 절대화 후 경로 불일치로 False(fail-safe).
      - 예외 발생 → False(fail-closed).
    """
    if not (isinstance(report_path, str) and report_path.strip()):
        return False
    if not (isinstance(result_path, str) and result_path.strip()):
        return False
    rp = report_path.strip()
    res = result_path.strip()
    try:
        # 상대경로 절대화에 쓸 공통 anchor(root 제공 시 우선, 그다음 cwd).
        base_anchors = []
        if isinstance(root, str) and root.strip():
            base_anchors.append(os.path.abspath(root.strip()))
        base_anchors.append(os.getcwd())

        def _candidates(path, extra_anchors=()):
            # absolute-normalized 후보 집합(symlink 미해소).
            if os.path.isabs(path):
                return {os.path.normpath(path)}
            out = set()
            for a in list(base_anchors) + list(extra_anchors):
                out.add(os.path.normpath(os.path.abspath(os.path.join(a, path))))
            return out

        res_cands = _candidates(res)
        # report 가 result.json 디렉토리 기준 basename 상대경로인 경우도 허용.
        res_dirs = {os.path.dirname(c) for c in res_cands if os.path.dirname(c)}
        rep_cands = _candidates(rp, res_dirs)
        return bool(res_cands & rep_cands)
    except Exception:
        return False


# ── provenance 검증 (14조건, 순서대로) ────────────────────────────────────────
def validate_executor_provenance(
    result: object,  # untrusted input — dict 여부를 조건5 에서 hard-fail 검사(타입 가정 금지)
    result_path: str,
    *,
    root: str,
    capability_reader=None,           # default: memory/capabilities/<task_id>.json 읽어 dict 반환. None이면 기본 reader 사용
    activation_epoch: Optional[float] = None,
    ledger_reader=None,               # callable(task_id, sha256)->bool, True면 중복. None이면 중복검사 skip(False 취급)
    forbidden_key_literals: tuple = (),  # 조건14: 이 literal들이 result에 있으면 quarantine. 비어있으면 scan skip
    stat_fn=None,                     # default os.stat (테스트 주입용)
    hash_fn=None,                     # default hashlib.sha256 기반 파일 해시 (테스트 주입용)
) -> ProvenanceResult:
    """envelope 부재 executor_result 의 provenance 를 14조건 hard-fail 검증.

    검사 순서: 1→2→3→4→5→6→7→8(sha계산)→9→10→11→12→13→14.
    전부 통과 → ProvenanceResult(ok=True, source_result_sha256=<sha>).
    """
    cap_reader = capability_reader or (lambda tid: _default_capability_reader(tid, root))
    hash_fn = hash_fn or _default_hash_fn
    stat_fn = stat_fn or os.stat

    # ── 조건5(선행 일부): result dict + task_id str 재확인 ────────────────────
    #   (task_id 는 조건1/2 에서 reader 호출/비교에 필요하므로 먼저 type 확인.)
    if not isinstance(result, dict):
        return ProvenanceResult(ok=False, reason=REASON_SCHEMA_FAIL)
    task_id = result.get("task_id")
    if not (isinstance(task_id, str) and task_id.strip()):
        return ProvenanceResult(ok=False, reason=REASON_SCHEMA_FAIL)
    task_id = task_id.strip()

    # ── 조건0(보안): task_id path traversal 방지 — capability path 구성 전 whitelist 검증 ──
    #   주입된 capability_reader 와 무관하게 항상 적용(보안 reason 을 surface).
    if not _validate_task_id_safe(task_id):
        return ProvenanceResult(ok=False, reason=REASON_TASK_ID_PATH_TRAVERSAL)

    # ── 조건1: dispatch record 존재 ───────────────────────────────────────────
    try:
        snapshot = cap_reader(task_id)
    except Exception:  # noqa: BLE001 — reader 실패 → 기록 미존재로 보수 처리
        snapshot = None
    if not isinstance(snapshot, dict):
        return ProvenanceResult(ok=False, reason=REASON_NO_DISPATCH_RECORD)

    # ── 조건2: task_id 일치 ───────────────────────────────────────────────────
    if result.get("task_id") != snapshot.get("task_id"):
        return ProvenanceResult(ok=False, reason=REASON_TASK_ID_MISMATCH)

    # ── 조건3: capsule_sha 일치 ───────────────────────────────────────────────
    capsule_sha = result.get("capsule_sha256")
    snap_sha = snapshot.get("source_sha256")
    if not (isinstance(capsule_sha, str) and capsule_sha):
        return ProvenanceResult(ok=False, reason=REASON_CAPSULE_SHA_MISMATCH)
    if capsule_sha != snap_sha:
        return ProvenanceResult(ok=False, reason=REASON_CAPSULE_SHA_MISMATCH)

    # ── 조건4: source path ∈ inbox ────────────────────────────────────────────
    #   result_path 정규화 == <root>/memory/events/p0b_inbox/<task_id>.result.json
    expected = _normalize(os.path.join(root, INBOX_REL, f"{task_id}.result.json"))
    actual = _normalize(result_path)
    if actual != expected:
        # basename + 부모 디렉토리(inbox) 확인 (정규화 후 동일성 외 추가 방어).
        return ProvenanceResult(ok=False, reason=REASON_SOURCE_PATH_INVALID)

    # ── 조건5: schema 유효 (report_path 필수) — task-2755 Wiring2 schema contract 정렬 ──
    #   report_path 는 (a) result.json 자기 자신(self-ref) 또는 (b) 실제 존재 report 파일,
    #   또는 (c) 그 외 비어있지 않은 report 참조 문자열 을 **둘 다(모두) 허용**한다.
    #   result-only task 는 별도 report 파일 생성 없이 self-ref 로 schema 를 만족할 수 있다.
    #   missing/빈 report_path 만 명확한 schema_fail(조용한 통과 0).
    report_path = result.get("report_path")
    if not (isinstance(report_path, str) and report_path.strip()):
        return ProvenanceResult(ok=False, reason=REASON_SCHEMA_FAIL)
    # self-ref(report_path_self_ref) / 실제 존재 report 파일 / 그 외 비어있지 않은
    #   report 참조 str 은 모두 schema 를 통과합니다(둘 다 허용). 추가 거부·디스크 I/O 0.
    #   self-ref 판정은 report_path_self_ref() helper 로 호출자/테스트가 명시 확인합니다.

    # ── 조건6: terminal_status 유효 ───────────────────────────────────────────
    if result.get("terminal_status") not in VALID_TERMINAL_STATUS:
        return ProvenanceResult(ok=False, reason=REASON_TERMINAL_INVALID)

    # ── 조건7: completion_signal 유효 ─────────────────────────────────────────
    if result.get("completion_signal") != EXECUTOR_COMPLETION_SIGNAL:
        return ProvenanceResult(ok=False, reason=REASON_COMPLETION_INVALID)

    # ── 조건8: result sha256 계산 ─────────────────────────────────────────────
    try:
        sha = hash_fn(result_path)
    except OSError:
        # 파일 read 불가 → schema/source 신뢰 불가 → source_path_invalid 보수 처리.
        return ProvenanceResult(ok=False, reason=REASON_SOURCE_PATH_INVALID)
    sha = str(sha)

    # ── 조건9: epoch 이후 result ───────────────────────────────────────────────
    if activation_epoch is not None:
        try:
            st = stat_fn(result_path)
            mtime = float(st.st_mtime)
        except (OSError, AttributeError, ValueError):
            # stat 실패 → 안전 분류 불가 → stale 취급(skip, quarantine 아님).
            return ProvenanceResult(
                ok=False, skip=True,
                reason=REASON_PRE_ACTIVATION_EPOCH,
                source_result_sha256=sha,
            )
        if mtime < float(activation_epoch):
            return ProvenanceResult(
                ok=False, skip=True,
                reason=REASON_PRE_ACTIVATION_EPOCH,
                source_result_sha256=sha,
            )

    # ── 조건10: duplicate / idempotency ───────────────────────────────────────
    if ledger_reader is not None:
        try:
            dup = bool(ledger_reader(task_id, sha))
        except Exception:  # noqa: BLE001 — ledger 조회 실패 → 비중복 취급(fail-open dedupe 금지 아님: 재수거는 idempotent writer 가 흡수)
            dup = False
        if dup:
            return ProvenanceResult(
                ok=False, duplicate=True,
                reason=REASON_DUPLICATE,
                source_result_sha256=sha,
            )

    # ── 조건11: callback schedule claim 없음 ──────────────────────────────────
    if (
        bool(result.get("callback_schedule_created"))
        or bool(result.get("schedule_created_by_executor"))
        or bool(result.get("callback_fired_by_executor"))
    ):
        return ProvenanceResult(
            ok=False, reason=REASON_CALLBACK_CLAIM, source_result_sha256=sha,
        )

    # ── 조건12: self-collector claim 없음 ─────────────────────────────────────
    if bool(result.get("self_collector")):
        return ProvenanceResult(
            ok=False, reason=REASON_SELF_COLLECTOR_CLAIM, source_result_sha256=sha,
        )

    # ── 조건13: dev self-key claim 없음 ───────────────────────────────────────
    if bool(result.get("owner_key_proof_present")):
        return ProvenanceResult(
            ok=False, reason=REASON_SELF_KEY_CLAIM, source_result_sha256=sha,
        )

    # ── 조건14: raw key 없음 ───────────────────────────────────────────────────
    if forbidden_key_literals:
        try:
            blob = json.dumps(result, ensure_ascii=False)
        except (TypeError, ValueError):
            blob = str(result)
        for literal in forbidden_key_literals:
            if literal and str(literal) in blob:
                return ProvenanceResult(
                    ok=False, reason=REASON_RAW_KEY, source_result_sha256=sha,
                )

    # ── 전부 통과 → provenance PASS ───────────────────────────────────────────
    return ProvenanceResult(ok=True, source_result_sha256=sha)


__all__ = [
    "ProvenanceResult",
    "validate_executor_provenance",
    "report_path_self_ref",
    "REASON_NO_DISPATCH_RECORD",
    "REASON_TASK_ID_MISMATCH",
    "REASON_CAPSULE_SHA_MISMATCH",
    "REASON_SOURCE_PATH_INVALID",
    "REASON_SCHEMA_FAIL",
    "REASON_TERMINAL_INVALID",
    "REASON_COMPLETION_INVALID",
    "REASON_PRE_ACTIVATION_EPOCH",
    "REASON_DUPLICATE",
    "REASON_CALLBACK_CLAIM",
    "REASON_SELF_COLLECTOR_CLAIM",
    "REASON_SELF_KEY_CLAIM",
    "REASON_RAW_KEY",
    "REASON_TASK_ID_PATH_TRAVERSAL",
    "VALID_TERMINAL_STATUS",
    "EXECUTOR_COMPLETION_SIGNAL",
]