import base64
import logging
import os
import re
import shutil
from pathlib import Path
from tempfile import NamedTemporaryFile
from typing import Iterable

from fastapi import HTTPException
from passlib.context import CryptContext

from .settings import get_settings

logger = logging.getLogger(__name__)

# Use argon2id for new hashes; keep bcrypt/pbkdf2 for legacy verification.
pwd_context = CryptContext(
    bcrypt__truncate_error=True,
    schemes=["argon2", "bcrypt", "pbkdf2_sha512"],
    deprecated="auto",
    default="argon2",
    argon2__type="ID",
)
DOMAIN_RE = re.compile(
    r"^(?=.{1,253}$)(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$"
)
LOCAL_PART_RE = re.compile(r"^(?=.{1,64}$)[A-Za-z0-9.!#$%&'*+/=?^_`{|}~-]+$")
SELECTOR_RE = re.compile(r"^[A-Za-z0-9_-]{1,63}$")


def hash_password(password: str) -> str:
    return pwd_context.hash(password)


def _mail_home() -> Path | None:
    """Return configured mail home if available, otherwise None."""
    try:
        settings = get_settings()
        mail_home = getattr(settings, "mail_home", None)
        if not mail_home:
            return None
        return Path(mail_home).absolute()
    except Exception as exc:
        logger.warning("_mail_home: suppressed %s: %s", type(exc).__name__, exc)
        return None


def mailbox_path_for(domain: str, local_part: str) -> Path:
    """Return the absolute Maildir path for a virtual mailbox."""
    base = _mail_home()
    if base is None:
        raise ValueError("Mail home is not configured")
    safe_domain = normalize_domain(domain)
    safe_local = validate_local_part(local_part)
    path = base / safe_domain / safe_local
    # A contained link can point at a *different* mailbox just as dangerously as
    # an escaping link. Preserve the lexical identity for descriptor-based I/O.
    if (base / safe_domain).is_symlink() or path.is_symlink():
        raise ValueError("Symlinked mailbox paths are not allowed")
    return path


def remove_mailbox(domain: str, local_part: str) -> bool:
    """Best-effort removal of a virtual mailbox directory under mail_home."""
    try:
        path = mailbox_path_for(domain, local_part)
    except ValueError:
        return False
    if not path.exists():
        return False
    try:
        from .file_safety import directory_fd

        with directory_fd(path.parent) as parent_fd:
            shutil.rmtree(path.name, dir_fd=parent_fd)
        logger.info("Removed mailbox directory %s", path)
        return True
    except OSError as exc:
        logger.warning("Unable to remove mailbox directory %s in-process: %s; trying privileged helper", path, exc)
        try:
            from .admin_ops import run_root_script

            run_root_script("manage-mailbox-fs.sh", "delete-maildir", domain, local_part)
            logger.info("Removed mailbox directory %s via root helper", path)
            return True
        except Exception as helper_exc:
            logger.warning("Unable to remove mailbox directory %s via root helper: %s", path, helper_exc)
            return False


def remove_domain_mailboxes(domain: str) -> bool:
    """Best-effort removal of an entire domain mail tree under mail_home."""
    base = _mail_home()
    if base is None:
        return False
    try:
        safe_domain = normalize_domain(domain)
    except ValueError:
        return False
    path = base / safe_domain
    if path.is_symlink():
        return False
    if path == base or not path.exists():
        return False
    try:
        from .file_safety import directory_fd

        with directory_fd(base) as parent_fd:
            shutil.rmtree(path.name, dir_fd=parent_fd)
        logger.info("Removed domain mailbox tree %s", path)
        return True
    except OSError as exc:
        logger.warning("Unable to remove domain mailbox tree %s in-process: %s; trying privileged helper", path, exc)
        try:
            from .admin_ops import run_root_script

            run_root_script("manage-mailbox-fs.sh", "delete-domain", domain)
            logger.info("Removed domain mailbox tree %s via root helper", path)
            return True
        except Exception as helper_exc:
            logger.warning("Unable to remove domain mailbox tree %s via root helper: %s", path, helper_exc)
            return False


def verify_password(password: str, hashed: str) -> bool:
    # bcrypt raises on >72-byte secrets; truncate for legacy bcrypt hashes.
    if hashed.startswith("$2"):
        secret = password.encode("utf-8")
        if len(secret) > 72:
            password = secret[:72].decode("utf-8", errors="ignore")
    return pwd_context.verify(password, hashed)


def normalize_domain(value: str) -> str:
    candidate = value.strip().rstrip(".").lower()
    try:
        normalized = candidate.encode("idna").decode("ascii")
    except UnicodeError as exc:
        raise ValueError("Invalid domain name") from exc
    if not DOMAIN_RE.fullmatch(normalized):
        raise ValueError("Invalid domain name")
    return normalized


def validate_local_part(value: str) -> str:
    candidate = value.strip()
    if not LOCAL_PART_RE.fullmatch(candidate) or "/" in candidate or "\\" in candidate or candidate.startswith(".") or candidate.endswith(".") or ".." in candidate:
        raise ValueError("Invalid local part")
    return candidate.lower()


def validate_selector(value: str) -> str:
    candidate = value.strip().lower()
    if not SELECTOR_RE.fullmatch(candidate):
        raise ValueError("Invalid DKIM selector")
    return candidate


def build_dkim_key_path(domain: str, selector: str, base_dir: Path) -> Path:
    try:
        safe_domain = normalize_domain(domain)
        safe_selector = validate_selector(selector)
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc
    base_path = base_dir.resolve(strict=False)
    key_path = (base_path / f"{safe_domain}.{safe_selector}.key").resolve(strict=False)
    try:
        key_path.relative_to(base_path)
    except ValueError as exc:
        raise HTTPException(status_code=400, detail="Invalid DKIM path") from exc
    return key_path


def remove_managed_dkim_key(raw_path: str | Path | None) -> bool:
    if not raw_path:
        return False
    base_dir = get_settings().dkim_keys_dir.resolve(strict=False)
    candidate = Path(raw_path).resolve(strict=False)
    try:
        candidate.relative_to(base_dir)
    except ValueError:
        return False
    if candidate.suffix != ".key":
        return False
    try:
        candidate.unlink(missing_ok=True)
    except OSError:
        return False
    return True


def _atomic_write(path: Path, content: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with NamedTemporaryFile("w", encoding="utf-8", dir=path.parent, delete=False) as tmp_file:
        tmp_file.write(content)
        tmp_path = Path(tmp_file.name)
    os.replace(tmp_path, path)
    path.chmod(0o640)


def sync_dkim_signing_maps(domains: Iterable[object]) -> None:
    """Write Rspamd selector/path maps for active domain keys.

    Only the *current* selector is used for signing. Previous keys are kept on
    disk/DNS for verification during the configured overlap window, but are not
    used for new signatures.
    """
    settings = get_settings()
    base_dir = settings.dkim_keys_dir.resolve(strict=False)
    selector_lines: list[str] = []
    path_lines: list[str] = []

    for domain in domains:
        raw_path = getattr(domain, "dkim_private_path", None)
        if not raw_path:
            continue
        safe_domain = normalize_domain(getattr(domain, "name", ""))
        safe_selector = validate_selector(getattr(domain, "dkim_selector", "default") or "default")
        key_path = Path(str(raw_path)).resolve(strict=False)
        try:
            key_path.relative_to(base_dir)
        except ValueError:
            continue
        selector_lines.append(f"{safe_domain} {safe_selector}")
        path_lines.append(f"{safe_domain} {key_path}")

    if not selector_lines and not path_lines and not base_dir.exists():
        return

    _atomic_write(base_dir / "selectors.map", "\n".join(selector_lines) + ("\n" if selector_lines else ""))
    _atomic_write(base_dir / "paths.map", "\n".join(path_lines) + ("\n" if path_lines else ""))


DKIM_KEY_BITS = 2048


def generate_dkim_key(domain: str, selector: str = "default") -> dict:
    """Create the DKIM keypair for a domain and return its public DNS record.

    Generated in-process with ``cryptography`` rather than by shelling out to
    ``rspamadm dkim_keygen``. Producing an RSA keypair is pure crypto and does not
    justify a hard dependency on the spam filter: with the old shell-out, adding a
    domain failed outright on installs configured with ENABLE_RSPAMD=no, and every
    test touching domain creation needed rspamd present on the machine.

    The private key is written as PKCS#8 PEM, the same format rspamd reads.
    """
    from cryptography.hazmat.primitives import serialization
    from cryptography.hazmat.primitives.asymmetric import rsa

    settings = get_settings()
    try:
        safe_domain = normalize_domain(domain)
        safe_selector = validate_selector(selector)
        key_path = build_dkim_key_path(safe_domain, safe_selector, settings.dkim_keys_dir)
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc)) from exc

    key_path.parent.mkdir(parents=True, exist_ok=True)
    try:
        private_key = rsa.generate_private_key(public_exponent=65537, key_size=DKIM_KEY_BITS)
        private_pem = private_key.private_bytes(
            encoding=serialization.Encoding.PEM,
            format=serialization.PrivateFormat.PKCS8,
            encryption_algorithm=serialization.NoEncryption(),
        )
        public_der = private_key.public_key().public_bytes(
            encoding=serialization.Encoding.DER,
            format=serialization.PublicFormat.SubjectPublicKeyInfo,
        )
    except Exception as exc:  # pragma: no cover - key generation failing is fatal anyway
        logger.error("DKIM key generation failed for %s: %s", safe_domain, exc)
        raise HTTPException(status_code=500, detail=f"DKIM key generation failed: {exc}") from exc

    # Write through a private temp file so a reader never sees a partial key.
    tmp_path = key_path.with_name(key_path.name + ".tmp")
    try:
        fd = os.open(tmp_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
        try:
            os.write(fd, private_pem)
        finally:
            os.close(fd)
        os.replace(tmp_path, key_path)
        # Group-readable so the rspamd service user (member of mailkeys) can sign.
        key_path.chmod(0o640)
    except OSError as exc:
        try:
            tmp_path.unlink(missing_ok=True)
        except OSError:
            pass
        logger.error("Unable to store DKIM private key at %s: %s", key_path, exc)
        raise HTTPException(status_code=500, detail=f"Unable to store DKIM private key: {exc}") from exc

    public_b64 = base64.b64encode(public_der).decode("ascii")
    dns_record = f"v=DKIM1; k=rsa; p={public_b64}"
    return {"path": str(key_path), "dns": dns_record, "selector": safe_selector}


def timestamp_dkim_selector(prefix: str = "s") -> str:
    """Build a unique DKIM selector suitable for scheduled rotations."""
    from datetime import UTC, datetime

    stamp = datetime.now(UTC).strftime("%Y%m%d%H%M")
    candidate = f"{prefix}{stamp}"
    return validate_selector(candidate[:63])


def default_dkim_rotation_interval_days() -> int:
    import os

    try:
        return max(0, int(os.getenv("LIMRISTEM_MAIL_DKIM_ROTATION_INTERVAL_DAYS", "90")))
    except ValueError:
        return 90


def default_dkim_overlap_days() -> int:
    import os

    try:
        return max(0, int(os.getenv("LIMRISTEM_MAIL_DKIM_OVERLAP_DAYS", "14")))
    except ValueError:
        return 14


def resolve_domain_rotation_interval_days(domain: object) -> int:
    value = int(getattr(domain, "dkim_rotation_interval_days", 0) or 0)
    return value if value > 0 else default_dkim_rotation_interval_days()


def resolve_domain_overlap_days(domain: object) -> int:
    value = int(getattr(domain, "dkim_overlap_days", 0) or 0)
    return value if value > 0 else default_dkim_overlap_days()