"""Local encrypted vault for browser autofill secrets.

Profile-scoped, model-blind credential store. Metadata (kind, label, origin,
timestamps) lives alongside an encrypted secret payload; the payload is
encrypted at rest with a locally generated Fernet key. The model only ever
sees opaque handles + metadata — secret values are resolved server-side by
the browser fill path and never enter tool results, logs, or the session DB.

Design notes:
- Follows the repo's "default frictionless, 0600 files OK" policy: the key
  file and vault file are created 0600 under ``<HERMES_HOME>/vault/``.
- Ported design (opaque-handle vault fill) from Merit-Systems/OpenInstinct
  (MIT): lib/manager/server/secret-store.ts + vault services.
- Three item kinds: ``login`` (password-only secret), ``payment`` (card fields) and
  ``address``; ``PAYMENT_FIELDS`` / ``ADDRESS_FIELDS`` are the canonical payload names.
"""

from __future__ import annotations

import json
import os
import re
import threading
import uuid
from contextlib import contextmanager, suppress
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, List, Optional
from urllib.parse import urlsplit

from hermes_constants import get_hermes_home
from utils import atomic_write_bytes

VAULT_KINDS = ("login", "payment", "address")

LOGIN_IDENTIFIER_TYPES = ("email", "phone", "username")

# Canonical secret-payload fields per non-login kind. Each maps to the WHATWG autocomplete token the
# browser fill targets (agent/vault_login_classifier.py); the Desktop Add dialog and `hermes vault add`
# both write these names, so the fill never has to guess a user's ad-hoc field naming.
PAYMENT_FIELDS = {
    "card_number": "cc-number", "cardholder_name": "cc-name", "exp_month": "cc-exp-month",
    "exp_year": "cc-exp-year", "cvc": "cc-csc", "billing_postal_code": "postal-code",
}
ADDRESS_FIELDS = {
    "address_line1": "address-line1", "address_line2": "address-line2", "city": "address-level2",
    "state": "address-level1", "postal_code": "postal-code", "country": "country-name",
}
REQUIRED_FIELDS = {"payment": ("card_number", "exp_month", "exp_year", "cvc"),
                   "address": ("address_line1", "city", "postal_code", "country")}

_DEFAULT_PORTS = {"http": 80, "https": 443}

_LOCK = threading.Lock()

# fcntl is Unix-only; Windows locks a byte range with msvcrt (same shape as tools/skill_usage.py).
msvcrt = None
try:
    import fcntl
except ImportError:  # pragma: no cover - platform-specific fallback
    fcntl = None
    with suppress(ImportError):
        import msvcrt


class VaultError(Exception):
    """Vault failure that is safe to surface (never contains secret values)."""


_OTP_ALGOS = {"SHA1": "sha1", "SHA256": "sha256", "SHA512": "sha512"}


def normalize_otp_secret(value: str) -> str:
    """Accept a raw base32 seed or an ``otpauth://totp/...`` URI. Returns the canonical stored form:
    the bare uppercase base32 seed, followed by ``|digits|period|algo`` ONLY when the URI departs from
    the RFC 6238 defaults (6 / 30 / SHA1), so a plain seed stays a plain seed. Non-default parameters
    are honoured, not dropped: an 8-digit or 60-second authenticator would otherwise get wrong codes."""
    value = (value or "").strip()
    if not value:
        return ""
    digits, period, algo = 6, 30, "SHA1"
    if value.lower().startswith("otpauth://"):
        from urllib.parse import parse_qs, urlparse
        parsed = urlparse(value)
        if parsed.netloc.lower() != "totp":
            raise VaultError("only otpauth://totp links are supported (counter-based HOTP is not)")
        qs = {k.lower(): v[0] for k, v in parse_qs(parsed.query).items()}
        value = qs.get("secret", "")
        try:
            digits = int(qs.get("digits", digits))
            period = int(qs.get("period", period))
        except ValueError:
            raise VaultError("otpauth:// digits/period must be integers")
        algo = qs.get("algorithm", algo).upper().replace("-", "")
        if digits not in (6, 7, 8) or period <= 0 or algo not in _OTP_ALGOS:
            raise VaultError("unsupported otpauth:// parameters (digits 6-8, period > 0, SHA1/SHA256/SHA512)")
    seed = re.sub(r"[\s-]", "", value).upper().rstrip("=")
    if not seed or re.search(r"[^A-Z2-7]", seed):
        raise VaultError("authenticator key must be a base32 secret or an otpauth:// URI")
    if (digits, period, algo) == (6, 30, "SHA1"):
        return seed
    return f"{seed}|{digits}|{period}|{algo}"


def totp_now(seed: str, *, digits: int = 6, period: int = 30, at: Optional[float] = None) -> str:
    """RFC 6238 TOTP for a stored seed (see normalize_otp_secret for the ``seed|digits|period|algo``
    form). Stdlib only."""
    import base64
    import hashlib
    import hmac
    import struct
    import time as _time

    algo = "sha1"
    if "|" in seed:
        seed, d, p, a = seed.split("|", 3)
        digits, period, algo = int(d), int(p), _OTP_ALGOS.get(a.upper(), "sha1")
    key = base64.b32decode(seed + "=" * (-len(seed) % 8), casefold=True)
    counter = int((at if at is not None else _time.time()) // period)
    digest = hmac.new(key, struct.pack(">Q", counter), getattr(hashlib, algo)).digest()
    offset = digest[-1] & 0x0F
    code = (struct.unpack(">I", digest[offset:offset + 4])[0] & 0x7FFFFFFF) % (10 ** digits)
    return str(code).zfill(digits)


def normalize_origin(url_or_origin: str) -> str:
    """Normalize a URL or origin to ``scheme://host[:port]``.

    Default ports (80 for http, 443 for https) are stripped so that
    ``https://example.com`` and ``https://example.com:443`` compare equal.
    Raises :class:`VaultError` for values without a scheme + host.
    """
    value = (url_or_origin or "").strip()
    if not value:
        raise VaultError("origin is required")
    if "://" not in value:
        raise VaultError(f"origin must include a scheme (got {value!r})")
    parts = urlsplit(value)
    scheme = (parts.scheme or "").lower()
    host = (parts.hostname or "").lower()
    if not scheme or not host:
        raise VaultError(f"could not parse origin from {value!r}")
    try:
        port = parts.port
    except ValueError as exc:
        raise VaultError(f"invalid port in origin {value!r}") from exc
    if port is None or port == _DEFAULT_PORTS.get(scheme):
        return f"{scheme}://{host}"
    return f"{scheme}://{host}:{port}"


@dataclass(frozen=True)
class VaultItemMeta:
    """Metadata-only view of a vault item. Never contains secret values.

    For ``kind='login'`` the identifier (email/username/phone) is metadata,
    not a secret: the agent may see it and type it itself. Only the password
    is vault-secret.
    """

    id: str
    kind: str
    label: str
    origin: Optional[str]
    created_at: str
    identifier_type: Optional[str] = None
    identifier: Optional[str] = None
    has_otp: bool = False  # a TOTP seed is stored: 2FA codes can be minted without asking the user
    # Every origin the password manager bound to this item (manager backends only;
    # ``origin`` is the first/primary one). Fill matching stays exact-origin against
    # this list — no wildcard or subdomain inference is ever derived from it.
    allowed_origins: tuple = ()

    def to_dict(self) -> Dict[str, Any]:
        out = {
            "id": self.id,
            "kind": self.kind,
            "label": self.label,
            "origin": self.origin,
            "created_at": self.created_at,
        }
        if self.identifier is not None:
            out["identifier"] = self.identifier
            out["identifier_type"] = self.identifier_type
        if self.has_otp:
            out["has_otp"] = True
        if len(self.allowed_origins) > 1:
            out["allowed_origins"] = list(self.allowed_origins)
        return out


class VaultStore:
    """Encrypted, profile-scoped vault under ``<HERMES_HOME>/vault/``."""

    def __init__(self, base_dir: Optional[Path] = None):
        self._base = Path(base_dir) if base_dir is not None else (
            Path(get_hermes_home()) / "vault"
        )
        self._vault_path = self._base / "vault.json.enc"
        self._key_path = self._base / "vault.key"

    # -- key / crypto ------------------------------------------------------

    def _ensure_dir(self) -> None:
        self._base.mkdir(mode=0o700, parents=True, exist_ok=True)
        # Route through the canonical securer (honors managed/NixOS
        # group-share mode and HERMES_UID/GID ownership) rather than a
        # bespoke chmod — same requirement as the browser-profile snapshot
        # dir (f1d05c review).
        try:
            from hermes_cli.config import _secure_dir

            _secure_dir(self._base)
        except Exception:
            try:
                os.chmod(self._base, 0o700)
            except OSError:
                pass

    @contextmanager
    def _locked(self):
        """Serialize read-modify-write cycles across threads AND processes: the Desktop gateway, a CLI
        `hermes vault add` and a TUI slash worker all write the same ``vault.json.enc``; two unlocked
        writers would drop each other's items."""
        with _LOCK:
            self._ensure_dir()
            lock_path = self._base / ".vault.lock"
            if msvcrt and (not lock_path.exists() or lock_path.stat().st_size == 0):
                lock_path.write_text(" ", encoding="utf-8")  # msvcrt needs a non-empty byte range to lock
            with open(lock_path, "r+" if msvcrt else "a+", encoding="utf-8") as fd:
                if fcntl:
                    fcntl.flock(fd, fcntl.LOCK_EX)
                elif msvcrt:
                    fd.seek(0)
                    msvcrt.locking(fd.fileno(), msvcrt.LK_LOCK, 1)
                try:
                    yield
                finally:
                    with suppress(OSError):
                        if fcntl:
                            fcntl.flock(fd, fcntl.LOCK_UN)
                        elif msvcrt:
                            fd.seek(0)
                            msvcrt.locking(fd.fileno(), msvcrt.LK_UNLCK, 1)

    def _fernet(self):
        from cryptography.fernet import Fernet

        self._ensure_dir()
        if not self._key_path.exists():
            key = Fernet.generate_key()
            fd = os.open(
                self._key_path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600
            )
            try:
                os.write(fd, key)
            finally:
                os.close(fd)
        else:
            key = self._key_path.read_bytes().strip()
        try:
            os.chmod(self._key_path, 0o600)
        except OSError:
            pass
        return Fernet(key)

    # -- persistence -------------------------------------------------------

    def _read_all(self) -> List[Dict[str, Any]]:
        if not self._vault_path.exists():
            return []
        blob = self._vault_path.read_bytes()
        if not blob:
            return []
        from cryptography.fernet import InvalidToken

        try:
            raw = self._fernet().decrypt(blob)
        except InvalidToken as exc:
            raise VaultError(
                "vault file could not be decrypted (key mismatch or corruption)"
            ) from exc
        data = json.loads(raw.decode("utf-8"))
        items = data.get("items", [])
        return items if isinstance(items, list) else []

    def _write_all(self, items: List[Dict[str, Any]]) -> None:
        self._ensure_dir()
        payload = json.dumps({"version": 1, "items": items}).encode("utf-8")
        blob = self._fernet().encrypt(payload)
        # fsync_dir: the directory entry must be durable too (power loss between rename and next sync).
        atomic_write_bytes(self._vault_path, blob, mode=0o600, fsync_dir=True)

    # -- public API ----------------------------------------------------------

    def add_item(
        self,
        kind: str,
        label: str,
        secret: Dict[str, Any],
        origin: Optional[str] = None,
    ) -> VaultItemMeta:
        """Add an item. ``secret`` is the sensitive payload (encrypted at rest).

        For ``kind='login'``, ``origin`` is required and the payload must
        contain ``identifier_type``, ``identifier`` and ``password``. The
        identifier fields are NOT secret — they are moved into item metadata
        (the agent may see and type the identifier itself); only
        ``password`` stays in the encrypted secret payload. ``payment`` and
        ``address`` payloads remain fully secret.
        """
        if kind not in VAULT_KINDS:
            raise VaultError(f"unknown vault kind {kind!r} (expected one of {VAULT_KINDS})")
        label = (label or "").strip()
        if not label:
            raise VaultError("label is required")
        norm_origin: Optional[str] = None
        identifier: Optional[str] = None
        identifier_type: Optional[str] = None
        secret = dict(secret)
        if kind == "login":
            if not origin:
                raise VaultError("origin is required for login items")
            norm_origin = normalize_origin(origin)
            id_type = secret.pop("identifier_type", None)
            if id_type not in LOGIN_IDENTIFIER_TYPES:
                raise VaultError(
                    f"identifier_type must be one of {LOGIN_IDENTIFIER_TYPES}"
                )
            identifier = str(secret.pop("identifier", "") or "").strip()
            if not identifier or not secret.get("password"):
                raise VaultError("login items require identifier and password")
            identifier_type = str(id_type)
            # Login secret payload is password (+ optional TOTP seed); identifier lives in
            # metadata and any stray origin echo is dropped.
            otp_secret = normalize_otp_secret(str(secret.get("otp_secret") or ""))
            secret = {"password": secret["password"], **({"otp_secret": otp_secret} if otp_secret else {})}
        else:
            allowed = PAYMENT_FIELDS if kind == "payment" else ADDRESS_FIELDS
            secret = {k: str(v) for k, v in secret.items() if k in allowed and str(v or "").strip()}
            missing = [f for f in REQUIRED_FIELDS[kind] if f not in secret]
            if missing:
                raise VaultError(f"{kind} items require {', '.join(missing)}")
            if origin:
                norm_origin = normalize_origin(origin)

        item_id = f"vault_{uuid.uuid4().hex[:12]}"
        record = {
            "id": item_id,
            "kind": kind,
            "label": label,
            "origin": norm_origin,
            "created_at": datetime.now(timezone.utc).isoformat(),
            "identifier_type": identifier_type,
            "identifier": identifier,
            "secret": dict(secret),
        }
        with self._locked():
            items = self._read_all()
            items.append(record)
            self._write_all(items)
        return self._meta(record)

    def list_items(self) -> List[VaultItemMeta]:
        """Metadata-only listing. Secret payloads are never included."""
        with self._locked():
            return [self._meta(rec) for rec in self._read_all()]

    def has_items(self) -> bool:
        try:
            with self._locked():
                return bool(self._read_all())
        except Exception:
            return False

    def remove_item(self, item_id: str) -> bool:
        with self._locked():
            items = self._read_all()
            remaining = [rec for rec in items if rec.get("id") != item_id]
            if len(remaining) == len(items):
                return False
            self._write_all(remaining)
            return True

    def get_meta(self, item_id: str) -> Optional[VaultItemMeta]:
        with self._locked():
            for rec in self._read_all():
                if rec.get("id") == item_id:
                    return self._meta(rec)
        return None

    def resolve_secret(self, item_id: str) -> Dict[str, Any]:
        """Resolve the decrypted secret payload for server-side use ONLY.

        Callers must never place the returned values into tool results,
        logs, exceptions, or any string that reaches the session DB.
        """
        with self._locked():
            for rec in self._read_all():
                if rec.get("id") == item_id:
                    return dict(rec.get("secret") or {})
        raise VaultError(f"no vault item with id {item_id!r}")

    @staticmethod
    def _meta(rec: Dict[str, Any]) -> VaultItemMeta:
        identifier = rec.get("identifier")
        return VaultItemMeta(
            id=str(rec.get("id", "")),
            kind=str(rec.get("kind", "")),
            label=str(rec.get("label", "")),
            origin=rec.get("origin"),
            created_at=str(rec.get("created_at", "")),
            identifier_type=rec.get("identifier_type") if identifier else None,
            identifier=identifier or None,
            has_otp=bool((rec.get("secret") or {}).get("otp_secret")),
        )


def get_vault_store() -> VaultStore:
    """Default profile-scoped vault store."""
    return VaultStore()


def scrub_secret_from_text(text: str, secret: Dict[str, Any]) -> str:
    """Defensively strip any secret values from a string (e.g. an exception
    message) before it can be surfaced. Case-sensitive exact substring scrub."""
    scrubbed = text
    for value in secret.values():
        if isinstance(value, str) and len(value) >= 3 and value in scrubbed:
            scrubbed = scrubbed.replace(value, "[REDACTED]")
    # Also collapse anything that looks like a leaked password-ish token in
    # common key=value echoes.
    scrubbed = re.sub(r"(password['\"]?\s*[:=]\s*)\S+", r"\1[REDACTED]", scrubbed)
    return scrubbed