"""Mailbox filesystem helpers: Maildir bootstrap, autoresponder sieve, SSRF-safe hosts."""

from __future__ import annotations

import ipaddress
import logging
import os
import re
import secrets
import socket
from datetime import datetime
from pathlib import Path
from typing import Any, Iterable

from sqlalchemy.orm import Session

from . import models
from .settings import get_settings
from .utils import hash_password, mailbox_path_for, normalize_domain, validate_local_part

logger = logging.getLogger(__name__)

# Bootstrap mailboxes created with every new domain.
DOMAIN_BOOTSTRAP_LOCAL_PARTS = ("postmaster", "abuse", "hostmaster")

# Private / special networks blocked for remote IMAP import (SSRF protection).
_BLOCKED_NETWORKS = (
    ipaddress.ip_network("0.0.0.0/8"),
    ipaddress.ip_network("10.0.0.0/8"),
    ipaddress.ip_network("100.64.0.0/10"),
    ipaddress.ip_network("127.0.0.0/8"),
    ipaddress.ip_network("169.254.0.0/16"),
    ipaddress.ip_network("172.16.0.0/12"),
    ipaddress.ip_network("192.0.0.0/24"),
    ipaddress.ip_network("192.0.2.0/24"),
    ipaddress.ip_network("192.168.0.0/16"),
    ipaddress.ip_network("198.18.0.0/15"),
    ipaddress.ip_network("198.51.100.0/24"),
    ipaddress.ip_network("203.0.113.0/24"),
    ipaddress.ip_network("224.0.0.0/4"),
    ipaddress.ip_network("240.0.0.0/4"),
    ipaddress.ip_network("::/128"),
    ipaddress.ip_network("::1/128"),
    ipaddress.ip_network("fc00::/7"),
    ipaddress.ip_network("fe80::/10"),
    ipaddress.ip_network("ff00::/8"),
    ipaddress.ip_network("2001:db8::/32"),
)

_SIEVE_SAFE_SUBJECT = re.compile(r'[\x00-\x1f\x7f"\\]')
_SIEVE_SAFE_BODY_CTRL = re.compile(r"[\x00-\x08\x0b\x0c\x0e-\x1f\x7f]")


def vmail_ids() -> tuple[int, int]:
    settings = get_settings()
    return int(settings.vmail_uid), int(settings.vmail_gid)


def import_allows_private_networks() -> bool:
    return os.getenv("LIMRISTEM_MAIL_IMPORT_ALLOW_PRIVATE_NETWORKS", "no").lower() in {
        "1",
        "true",
        "yes",
        "on",
    }


def _ip_is_blocked(address: ipaddress.IPv4Address | ipaddress.IPv6Address) -> bool:
    if address.is_unspecified or address.is_loopback or address.is_link_local or address.is_multicast:
        return True
    if address.is_private or address.is_reserved:
        return True
    return any(address in network for network in _BLOCKED_NETWORKS)


def _normalize_host_label(host: str) -> str:
    cleaned = (host or "").strip()
    if not cleaned:
        raise ValueError("Source host is required")
    if any(c in cleaned for c in (" ", "\t", "\r", "\n", "/", "\\", "?", "#")):
        raise ValueError("Invalid source host")
    if cleaned.startswith("[") and cleaned.endswith("]"):
        cleaned = cleaned[1:-1]
    return cleaned


def validate_remote_import_host(host: str, *, check_resolution: bool = True) -> str:
    """Validate IMAP source host against SSRF (no private/metadata targets by default).

    ``check_resolution=True`` (connect path) resolves DNS and rejects private/link-local
    addresses. Job creation may use ``check_resolution=False`` for syntax + literal checks only.
    """
    cleaned = _normalize_host_label(host)
    lowered = cleaned.lower().rstrip(".")
    if import_allows_private_networks():
        return cleaned
    if lowered in {"localhost", "localhost.localdomain", "metadata", "metadata.google.internal"}:
        raise ValueError("Source host resolves to a blocked network address")
    try:
        literal = ipaddress.ip_address(cleaned)
    except ValueError:
        literal = None
    if literal is not None and _ip_is_blocked(literal):
        raise ValueError("Source host resolves to a blocked network address")
    if not check_resolution:
        return cleaned
    resolve_allowed_import_addresses(cleaned)
    return cleaned


def resolve_allowed_import_addresses(host: str) -> list[str]:
    """Resolve host to non-blocked IP strings (SSRF allow-list for this connect)."""
    cleaned = _normalize_host_label(host)
    if import_allows_private_networks():
        try:
            infos = socket.getaddrinfo(cleaned, None, type=socket.SOCK_STREAM)
        except socket.gaierror as exc:
            raise ValueError(f"Unable to resolve source host: {cleaned}") from exc
        addrs: list[str] = []
        for info in infos:
            sockaddr = info[4]
            if sockaddr:
                addrs.append(str(sockaddr[0]))
        if not addrs:
            raise ValueError(f"Unable to resolve source host: {cleaned}")
        return addrs
    try:
        literal = ipaddress.ip_address(cleaned)
    except ValueError:
        literal = None
    if literal is not None:
        if _ip_is_blocked(literal):
            raise ValueError("Source host resolves to a blocked network address")
        return [str(literal)]
    try:
        infos = socket.getaddrinfo(cleaned, None, type=socket.SOCK_STREAM)
    except socket.gaierror as exc:
        raise ValueError(f"Unable to resolve source host: {cleaned}") from exc
    if not infos:
        raise ValueError(f"Unable to resolve source host: {cleaned}")
    allowed: list[str] = []
    for info in infos:
        sockaddr = info[4]
        if not sockaddr:
            continue
        try:
            addr = ipaddress.ip_address(sockaddr[0])
        except ValueError:
            continue
        if _ip_is_blocked(addr):
            raise ValueError("Source host resolves to a blocked network address")
        allowed.append(str(addr))
    if not allowed:
        raise ValueError(f"Unable to resolve source host: {cleaned}")
    return allowed


def assert_peer_address_allowed(peer: str, *, allowed: list[str] | None = None) -> None:
    """Reject connections whose peer IP is private/blocked or not in the pre-resolved set."""
    try:
        addr = ipaddress.ip_address(peer)
    except ValueError as exc:
        raise ValueError(f"Invalid peer address: {peer}") from exc
    if not import_allows_private_networks() and _ip_is_blocked(addr):
        raise ValueError("Source host resolves to a blocked network address")
    if allowed is not None and not import_allows_private_networks():
        allowed_set = {str(ipaddress.ip_address(a)) for a in allowed}
        if str(addr) not in allowed_set:
            # Rebinding / TOCTOU: connect peer is not one of the IPs validated pre-connect.
            raise ValueError("Source peer address changed after resolution (possible DNS rebinding)")


def validate_public_https_url(url: str, *, allow_private: bool = False) -> str:
    """Validate an absolute HTTPS URL and reject credentials/query/fragment + private hosts."""
    from urllib.parse import urlparse

    raw = (url or "").strip()
    if not raw:
        raise ValueError("URL is required")
    parsed = urlparse(raw)
    if parsed.scheme != "https":
        raise ValueError("URL must use https://")
    if not parsed.hostname:
        raise ValueError("URL hostname is required")
    if parsed.username or parsed.password:
        raise ValueError("URL must not contain credentials")
    if parsed.query or parsed.fragment:
        raise ValueError("URL must not contain query or fragment")
    cleaned = _normalize_host_label(parsed.hostname)
    if not allow_private:
        lowered = cleaned.lower().rstrip(".")
        if lowered in {"localhost", "localhost.localdomain", "metadata", "metadata.google.internal"}:
            raise ValueError("URL host is not allowed")
        try:
            literal = ipaddress.ip_address(cleaned)
        except ValueError:
            literal = None
        if literal is not None:
            if _ip_is_blocked(literal):
                raise ValueError("URL host resolves to a blocked network address")
        else:
            try:
                infos = socket.getaddrinfo(cleaned, None, type=socket.SOCK_STREAM)
            except socket.gaierror as resolve_exc:
                raise ValueError(f"Unable to resolve URL host: {cleaned}") from resolve_exc
            for info in infos:
                sockaddr = info[4]
                if not sockaddr:
                    continue
                try:
                    addr = ipaddress.ip_address(sockaddr[0])
                except ValueError:
                    continue
                if _ip_is_blocked(addr):
                    raise ValueError("URL host resolves to a blocked network address")
    return raw.rstrip("/")


def _chown_chmod(path: Path, uid: int, gid: int, mode: int) -> None:
    """Best-effort ownership/mode.

    Apply mode *before* chown when possible: after CAP_CHOWN hands the tree to
    vmail, chmod needs CAP_FOWNER. Domain dirs must remain group-writable (2770).
    """
    try:
        os.chmod(path, mode)
    except OSError as exc:
        logger.debug("chmod(pre) %s -> %o failed: %s", path, mode, exc)
    try:
        os.chown(path, uid, gid)
    except OSError as exc:
        # Without CAP_CHOWN the service user cannot chown to vmail; group-write still works.
        logger.debug("chown %s -> %s:%s failed: %s", path, uid, gid, exc)
    try:
        # Re-apply after chown when CAP_FOWNER (or root) is available.
        os.chmod(path, mode)
    except OSError as exc:
        logger.debug("chmod(post) %s -> %o failed: %s", path, mode, exc)


def _maildir_ready(path: Path) -> bool:
    try:
        return all((path / sub).is_dir() for sub in ("cur", "new", "tmp"))
    except OSError:
        # vmail-owned 0700 trees are not stat-able by the service user.
        return False


def _mailbox_dir_exists_unreadable(path: Path) -> bool:
    """True when the mailbox directory name exists under a listable parent but is not enterable."""
    try:
        parent = path.parent
        if not parent.is_dir():
            return False
        return path.name in os.listdir(parent)
    except OSError:
        return False


def ensure_maildir_tree(path: Path, uid: int | None = None, gid: int | None = None) -> Path:
    """Create a Maildir (cur/new/tmp) with vmail ownership when possible.

    Create the full tree first, then chown. Early chown to vmail:700 would lock
    the service user (group vmail) out of creating special folders when CAP_CHOWN
    is available.

    Raises OSError if the essential INBOX folders cannot be created.
    """
    if uid is None or gid is None:
        uid, gid = vmail_ids()
    # Already a complete Maildir: either we can inspect cur/new/tmp, or the mailbox
    # dir exists as a private (0700) vmail tree that we must not re-enter.
    if _maildir_ready(path) or _mailbox_dir_exists_unreadable(path):
        try:
            parent = path.parent
            if parent.exists():
                os.chmod(parent, 0o2770)
                _chown_chmod(parent, uid, gid, 0o2770)
        except OSError:
            pass
        return path
    # Domain dir must stay group-writable (2770) so the API user can add mailboxes.
    parent = path.parent
    if parent.exists():
        try:
            os.chmod(parent, 0o2770)
        except OSError:
            pass
    try:
        path.mkdir(parents=True, exist_ok=True)
        for sub in ("cur", "new", "tmp"):
            (path / sub).mkdir(parents=True, exist_ok=True)
        # Special-use folders expected by Dovecot auto=subscribe.
        for special in ("Sent", "Drafts", "Junk", "Trash", "Archive"):
            special_path = path / f".{special}"
            special_path.mkdir(parents=True, exist_ok=True)
            for sub in ("cur", "new", "tmp"):
                (special_path / sub).mkdir(parents=True, exist_ok=True)
    except PermissionError:
        # Fall back to root helper when the service user cannot enter/create under vmail.
        from .admin_ops import run_root_script

        domain = path.parent.name
        local_part = path.name
        try:
            run_root_script("manage-mailbox-fs.sh", "ensure-maildir", domain, local_part)
        except Exception as exc:
            if _maildir_ready(path):
                return path
            raise OSError(f"Unable to create Maildir for {local_part}@{domain}: {exc}") from exc
        if _maildir_ready(path):
            return path
        raise
    # Seed subscriptions so clients see folders immediately (before chown).
    subs = path / "subscriptions"
    desired = ["Sent", "Drafts", "Junk", "Trash", "Archive"]
    existing: set[str] = set()
    if subs.exists():
        try:
            existing = {line.strip() for line in subs.read_text(encoding="utf-8").splitlines() if line.strip()}
        except OSError:
            existing = set()
    missing = [name for name in desired if name not in existing]
    if missing or not subs.exists():
        try:
            lines = sorted(existing.union(desired))
            subs.write_text("\n".join(lines) + "\n", encoding="utf-8")
        except OSError as exc:
            logger.warning("Unable to write subscriptions for %s: %s", path, exc)
    if not _maildir_ready(path):
        raise OSError(f"Maildir incomplete after create: {path}")
    # Ownership/mode last: mailbox private 700, domain dir setgid group-writable.
    # Walk bottom-up so we never chown a parent to vmail:700 before children
    # (which would lock the service user out of the tree mid-walk).
    try:
        for dirpath, _dirnames, filenames in os.walk(path, topdown=False):
            p = Path(dirpath)
            for name in filenames:
                _chown_chmod(p / name, uid, gid, 0o600)
            _chown_chmod(p, uid, gid, 0o700)
    except OSError as exc:
        logger.debug("walk chown for %s: %s", path, exc)
    try:
        parent = path.parent
        if parent.exists():
            _chown_chmod(parent, uid, gid, 0o2770)
    except OSError:
        pass
    return path


def _mailbox_fs_helper_candidates() -> list[Path]:
    settings = get_settings()
    base = Path(getattr(settings, "base_dir", "") or os.getenv("LIMRISTEM_MAIL_BASE_DIR", "") or "/opt/limristem-mail")
    return [
        base / "bin" / "manage-mailbox-fs.sh",
        Path(__file__).resolve().parent.parent / "scripts" / "manage-mailbox-fs.sh",
        Path("/opt/limristem-mail/bin/manage-mailbox-fs.sh"),
    ]


def _privileged_ensure_maildir(domain: str, local_part: str) -> Path | None:
    """Create Maildir via sudo helper (root → vmail ownership)."""
    import shutil
    import subprocess

    helper = next((p for p in _mailbox_fs_helper_candidates() if p.is_file() and os.access(p, os.X_OK)), None)
    if helper is None:
        return None
    cmd: list[str]
    if os.geteuid() == 0:
        cmd = [str(helper), "ensure-maildir", domain, local_part]
    else:
        sudo = shutil.which("sudo")
        if not sudo:
            return None
        cmd = [sudo, "-n", str(helper), "ensure-maildir", domain, local_part]
    try:
        result = subprocess.run(
            cmd,
            check=False,
            capture_output=True,
            text=True,
            timeout=30,
        )
    except (OSError, subprocess.SubprocessError) as exc:
        logger.warning("Privileged maildir helper failed for %s@%s: %s", local_part, domain, exc)
        return None
    if result.returncode != 0:
        err = (result.stderr or result.stdout or "").strip()
        logger.warning(
            "Privileged maildir helper exit %s for %s@%s: %s",
            result.returncode,
            local_part,
            domain,
            err,
        )
        return None
    path = mailbox_path_for(domain, local_part)
    if _maildir_ready(path):
        return path
    return None


def ensure_account_maildir(domain: str, local_part: str) -> Path:
    """Create Maildir immediately for CLI/API/panel/import account creation.

    Tries an in-process create first (works as root or when the service user is
    in group ``vmail`` with a group-writable mail home). Falls back to the
    passwordless sudo helper ``manage-mailbox-fs.sh`` so panel/API always get a
    real Maildir, not a lazy first-delivery path.
    """
    path = mailbox_path_for(domain, local_part)
    try:
        return ensure_maildir_tree(path)
    except OSError as direct_exc:
        logger.info(
            "Direct maildir create failed for %s@%s (%s); trying privileged helper",
            local_part,
            domain,
            direct_exc,
        )
        privileged = _privileged_ensure_maildir(domain, local_part)
        if privileged is not None:
            return privileged
        raise OSError(
            f"Unable to create Maildir for {local_part}@{domain}: {direct_exc}"
        ) from direct_exc


def generate_mailbox_password(length: int = 20) -> str:
    # URL-safe, high entropy; long enough for AccountCreate min length when used there.
    return secrets.token_urlsafe(max(16, length))


def bootstrap_domain_mailboxes(
    db: Session,
    domain: models.Domain,
    *,
    quota_mb: int = 2048,
    local_parts: Iterable[str] = DOMAIN_BOOTSTRAP_LOCAL_PARTS,
    password: str | None = None,
    create_catch_all: bool = True,
) -> dict[str, Any]:
    """Create primary role mailboxes + optional catch-all alias to postmaster."""
    created: list[dict[str, str]] = []
    skipped: list[str] = []
    shared_password = password  # if set, same for all bootstrap boxes
    for local in local_parts:
        try:
            safe_local = validate_local_part(local)
        except ValueError:
            continue
        email = f"{safe_local}@{domain.name}"
        existing = (
            db.query(models.Account)
            .filter(models.Account.domain_id == domain.id, models.Account.local_part == safe_local)
            .first()
        )
        if existing:
            skipped.append(email)
            # Still ensure Maildir exists (API/panel/import paths included).
            try:
                ensure_account_maildir(domain.name, safe_local)
            except Exception as exc:
                logger.error("Unable to ensure maildir for %s: %s", email, exc)
                raise
            continue
        plain = shared_password or generate_mailbox_password()
        account = models.Account(
            domain_id=domain.id,
            local_part=safe_local,
            username=email,
            password_hash=hash_password(plain),
            quota_mb=quota_mb,
            is_active=True,
        )
        db.add(account)
        db.flush()
        try:
            ensure_account_maildir(domain.name, safe_local)
        except Exception as exc:
            logger.error("Unable to create maildir for %s: %s", email, exc)
            raise
        created.append({"email": email, "password": plain, "local_part": safe_local})

    catch_all_created = False
    if create_catch_all:
        postmaster = f"postmaster@{domain.name}"
        existing_alias = (
            db.query(models.Alias)
            .filter(
                models.Alias.domain_id == domain.id,
                models.Alias.source_local == "*",
                models.Alias.destination == postmaster,
            )
            .first()
        )
        if not existing_alias:
            db.add(
                models.Alias(
                    domain_id=domain.id,
                    source_local="*",
                    destination=postmaster,
                    is_active=True,
                )
            )
            catch_all_created = True

    db.commit()
    return {
        "domain": domain.name,
        "created": created,
        "skipped": skipped,
        "catch_all": postmaster if create_catch_all else None,
        "catch_all_created": catch_all_created,
    }


def ensure_primary_hostname_domain(db: Session) -> models.Domain | None:
    """Ensure the server primary/hostname domain exists with bootstrap mailboxes."""
    settings = get_settings()
    candidates: list[str] = []
    for raw in (settings.primary_domain, settings.hostname):
        value = (raw or "").strip().rstrip(".").lower()
        if not value or "." not in value:
            continue
        try:
            candidates.append(normalize_domain(value))
        except ValueError:
            continue
    if not candidates:
        return None
    name = candidates[0]
    domain = db.query(models.Domain).filter(models.Domain.name == name).first()
    if domain:
        bootstrap_domain_mailboxes(db, domain)
        return domain
    domain = models.Domain(name=name, is_active=True, max_users=0, dmarc_policy="reject", dkim_selector="default")
    db.add(domain)
    db.commit()
    db.refresh(domain)
    bootstrap_domain_mailboxes(db, domain)
    return domain


def _sieve_quote(value: str) -> str:
    cleaned = _SIEVE_SAFE_BODY_CTRL.sub("", value or "")
    return cleaned.replace("\\", "\\\\").replace('"', '\\"')


def _sieve_subject(value: str) -> str:
    cleaned = _SIEVE_SAFE_SUBJECT.sub(" ", value or "Out of office")
    cleaned = re.sub(r"\s+", " ", cleaned).strip() or "Out of office"
    return cleaned[:200]


def build_vacation_sieve(
    *,
    subject: str,
    body: str,
    reply_once_days: int = 7,
    start_at: datetime | None = None,
    end_at: datetime | None = None,
) -> str:
    """Build a Sieve vacation script for the autoresponder."""
    days = max(1, min(int(reply_once_days or 7), 30))
    lines = [
        'require ["vacation", "date", "relational", "comparator-i;ascii-numeric"];',
        "",
    ]
    # Optional date window using currentdate (ISO date).
    if start_at is not None:
        start = start_at.strftime("%Y-%m-%d")
        lines.append(f'if currentdate :value "lt" "date" "{start}" {{ stop; }}')
    if end_at is not None:
        end = end_at.strftime("%Y-%m-%d")
        lines.append(f'if currentdate :value "gt" "date" "{end}" {{ stop; }}')
    if start_at is not None or end_at is not None:
        lines.append("")
    body_text = _sieve_quote(body or "I am currently unavailable.")
    subject_text = _sieve_subject(subject)
    lines.append("vacation")
    lines.append(f"  :days {days}")
    lines.append(f'  :subject "{subject_text}"')
    lines.append(f'  "{body_text}";')
    lines.append("")
    return "\n".join(lines)


def write_autoresponder_sieve(domain: str, local_part: str, script: str | None) -> Path:
    """Write or remove the active personal sieve script for a mailbox.

    Active path is ``~/sieve/active.sieve`` (not ``~/.dovecot.sieve``) so Maildir++
    does not treat the active script as a mailbox folder.
    """
    home = mailbox_path_for(domain, local_part)
    ensure_maildir_tree(home)
    uid, gid = vmail_ids()
    sieve_dir = home / "sieve"
    sieve_dir.mkdir(parents=True, exist_ok=True)
    try:
        os.chown(sieve_dir, uid, gid)
        os.chmod(sieve_dir, 0o700)
    except OSError:
        pass
    # Legacy path used before the maildir++ collision fix — always clean up.
    legacy_active = home / ".dovecot.sieve"
    active = sieve_dir / "active.sieve"
    script_path = sieve_dir / "autoresponder.sieve"
    if not script:
        for path in (script_path, active, legacy_active):
            try:
                if path.is_symlink() or path.exists():
                    path.unlink()
            except OSError:
                pass
        # Drop compiled binaries if present.
        for path in sieve_dir.glob("*.svbin"):
            try:
                path.unlink()
            except OSError:
                pass
        return home
    script_path.write_text(script, encoding="utf-8")
    try:
        os.chown(script_path, uid, gid)
        os.chmod(script_path, 0o600)
    except OSError:
        pass
    # Active script as symlink inside ~/sieve/ (matches templates/dovecot 90-sieve.conf).
    try:
        if active.exists() or active.is_symlink():
            active.unlink()
        if legacy_active.exists() or legacy_active.is_symlink():
            legacy_active.unlink()
        active.symlink_to("autoresponder.sieve")
        try:
            os.chown(active, uid, gid, follow_symlinks=False)
        except (OSError, TypeError, NotImplementedError):
            pass
    except OSError as exc:
        # Fallback: write content as regular file at the active path.
        logger.warning("Symlink active sieve failed for %s@%s: %s", local_part, domain, exc)
        active.write_text(script, encoding="utf-8")
        try:
            os.chown(active, uid, gid)
            os.chmod(active, 0o600)
        except OSError:
            pass
    return home


def apply_autoresponder_to_mailbox(account: models.Account, autoresponder: models.Autoresponder | None) -> None:
    domain = account.domain
    if not domain:
        return
    if not autoresponder or not autoresponder.is_enabled:
        write_autoresponder_sieve(domain.name, account.local_part, None)
        return
    script = build_vacation_sieve(
        subject=autoresponder.subject or "Out of office",
        body=autoresponder.body or "",
        reply_once_days=int(autoresponder.reply_once_days or 7),
        start_at=autoresponder.start_at,
        end_at=autoresponder.end_at,
    )
    write_autoresponder_sieve(domain.name, account.local_part, script)