#!/usr/bin/env python3
"""
connettore_metadataexporter
HTTP receiver for Rspamd metadata_exporter, compatibile con il database rqwatch.

Endpoints:
  POST /api/metadata_importer_multipart  — formatter = "multipart"  (raccomandata)
  POST /api/metadata_importer            — formatter = "default" + meta_headers
"""

import email as _email
import email.header
import email.policy
import json
import logging
import os
import re
import secrets
import uuid
from datetime import date
from pathlib import Path
from typing import Optional

import pymysql
from dotenv import load_dotenv
from fastapi import Depends, FastAPI, File, Form, HTTPException, Request, UploadFile
from fastapi.responses import PlainTextResponse
from fastapi.security import HTTPBasic, HTTPBasicCredentials
from pymysql.cursors import DictCursor

load_dotenv()

# ─── Configurazione ───────────────────────────────────────────────────────────

API_USER: str = os.getenv("RSPAMD_API_USER", "rspamd")
API_PASS: str = os.getenv("RSPAMD_API_PASS", "")
API_ACL: set[str] = {ip.strip() for ip in os.getenv("RSPAMD_API_ACL", "127.0.0.1").split(",")}
QUARANTINE_DIR: str = os.getenv("QUARANTINE_DIR", "/quarantine")
SERVER_ALIAS: str = os.getenv("MY_API_SERVER_ALIAS", "mx1")

DB_HOST: str = os.getenv("DB_HOST", "127.0.0.1")
DB_PORT: int = int(os.getenv("DB_PORT", "3306"))
DB_NAME: str = os.getenv("DB_NAME", "rqwatch")
DB_USER: str = os.getenv("DB_USER", "rqwatch")
DB_PASS: str = os.getenv("DB_PASS", "")
MAILLOGS_TABLE: str = os.getenv("MAILLOGS_TABLE", "mail_logs")
MAIL_RECIPIENTS_TABLE: str = os.getenv("MAIL_RECIPIENTS_TABLE", "mail_log_recipients")

_store_flags: dict[str, str] = {
    "no action":       os.getenv("STORE_NO_ACTION", "false"),
    "add header":      os.getenv("STORE_ADD_HEADER", "true"),
    "rewrite subject": os.getenv("STORE_REWRITE_SUBJECT", "true"),
    "greylist":        os.getenv("STORE_GREYLIST", "false"),
    "discard":         os.getenv("STORE_DISCARD", "true"),
    "reject":          os.getenv("STORE_REJECT", "true"),
}
STORE_ACTIONS: set[str] = {
    k for k, v in _store_flags.items()
    if v.strip().lower() in ("1", "true", "yes")
}

# Limiti di campo compatibili con rqwatch MailLog::FIELD_LIMITS
FIELD_LIMITS: dict[str, int] = {
    "qid": 30, "server": 10, "subject": 1024, "action": 20,
    "ip": 50, "mail_from": 255, "mime_from": 255,
    "rcpt_to": 1024, "mime_to": 1024, "mail_location": 255, "message_id": 1024,
}

# ─── Logging ──────────────────────────────────────────────────────────────────

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s [connettore] %(message)s",
)
log = logging.getLogger("connettore")

# ─── App ──────────────────────────────────────────────────────────────────────

app = FastAPI(docs_url=None, redoc_url=None)
_security = HTTPBasic()


# ─── Helper DB ────────────────────────────────────────────────────────────────

def _db() -> pymysql.Connection:
    return pymysql.connect(
        host=DB_HOST, port=DB_PORT, user=DB_USER, password=DB_PASS,
        database=DB_NAME, charset="utf8mb4", cursorclass=DictCursor,
        autocommit=False,
    )


def _insert(data: dict, recipients: list[str]) -> int:
    conn = _db()
    try:
        with conn.cursor() as cur:
            cols = ", ".join(f"`{k}`" for k in data)
            ph = ", ".join(["%s"] * len(data))
            cur.execute(
                f"INSERT INTO `{MAILLOGS_TABLE}` ({cols}) VALUES ({ph})",
                list(data.values()),
            )
            db_id: int = cur.lastrowid
            if db_id and recipients:
                unique = list({r.lower().strip() for r in recipients if r.strip()})
                cur.executemany(
                    f"INSERT INTO `{MAIL_RECIPIENTS_TABLE}` "
                    "(mail_log_id, recipient_email) VALUES (%s, %s)",
                    [(db_id, r) for r in unique],
                )
        conn.commit()
        return db_id
    except Exception:
        conn.rollback()
        raise
    finally:
        conn.close()


# ─── Helper email ─────────────────────────────────────────────────────────────

def _decode_header(value: str) -> str:
    try:
        parts = _email.header.decode_header(value)
        out = []
        for raw, charset in parts:
            if isinstance(raw, bytes):
                out.append(raw.decode(charset or "utf-8", errors="replace"))
            else:
                out.append(raw)
        return "".join(out)
    except Exception:
        return value


def _parse_mime(raw: bytes) -> dict:
    try:
        msg = _email.message_from_bytes(raw, policy=_email.policy.compat32)

        def hdr(name: str) -> str:
            v = msg.get(name, "")
            return _decode_header(str(v)) if v else ""

        raw_lines = [f"{k}: {v}" for k, v in msg.items()]
        headers_raw = "\r\n".join(raw_lines).encode("utf-8", "ignore").decode("utf-8")

        return {
            "mime_from":    hdr("From"),
            "mime_to":      hdr("To")[:1024],
            "mime_subject": hdr("Subject"),
            "message_id":   hdr("Message-ID"),
            "headers_raw":  headers_raw,
        }
    except Exception as exc:
        log.warning(f"MIME parse error: {exc}")
        return {
            "mime_from": "", "mime_to": "", "mime_subject": "",
            "message_id": "", "headers_raw": "",
        }


# ─── Helper quarantena ────────────────────────────────────────────────────────

def _store_email(qid: str, raw: bytes) -> Optional[str]:
    q = Path(QUARANTINE_DIR)
    if not q.is_dir() or not os.access(str(q), os.W_OK):
        log.error(f"Quarantine dir non accessibile: {QUARANTINE_DIR}")
        return None

    today = date.today().isoformat()
    safe = qid if re.match(r"^[a-zA-Z0-9]+$", qid) else ""
    subdir = safe if safe and safe != "unknown" else f"unknown/{uuid.uuid4().hex}"
    mail_dir = q / today / subdir
    mail_dir.mkdir(parents=True, exist_ok=True)

    dest = mail_dir / "mail.eml"
    try:
        dest.write_bytes(raw)
        return str(dest)
    except Exception as exc:
        log.error(f"Scrittura fallita {dest}: {exc}")
        return None


def _has_virus(symbols) -> bool:
    if isinstance(symbols, dict):
        return any(
            isinstance(v, dict) and v.get("group") == "antivirus"
            for v in symbols.values()
        )
    if isinstance(symbols, list):
        return any(
            isinstance(s, dict) and s.get("group") == "antivirus"
            for s in symbols
        )
    return False


def _trim_fields(data: dict) -> dict:
    for field, limit in FIELD_LIMITS.items():
        v = data.get(field)
        if isinstance(v, str) and len(v) > limit:
            log.warning(f"Campo '{field}' troncato a {limit} caratteri")
            data[field] = v[:limit]
    return data


def _sanitize_server(s: str) -> str:
    return re.sub(r"[^a-zA-Z0-9.\-]", "", s)[:10]


# ─── Logica principale ────────────────────────────────────────────────────────

def _process(
    *,
    qid: str,
    server: str,
    subject: str,
    score: float,
    action: str,
    symbols_json: str,
    fuzzy_json: str,
    ip: str,
    mail_from: str,
    rcpt_list: list[str],
    size: int,
    raw_email: bytes,
) -> int:
    try:
        symbols = json.loads(symbols_json)
    except Exception:
        symbols = {}

    virus = _has_virus(symbols)
    mail_stored, mail_location = 0, None

    if action in STORE_ACTIONS or virus:
        mail_location = _store_email(qid, raw_email)
        if mail_location:
            mail_stored = 1
            log.info(f"{qid} in quarantena: {mail_location}")
        else:
            log.error(f"{qid} salvataggio quarantena fallito")

    if not mail_from:
        mail_from = "empty-mail-from@localhost"

    mime = _parse_mime(raw_email)

    data = {
        "qid":           qid,
        "server":        _sanitize_server(server),
        "subject":       mime["mime_subject"] or subject or "",
        "score":         round(score, 2),
        "action":        action,
        "symbols":       symbols_json or "[]",
        "has_virus":     1 if virus else 0,
        "fuzzy_hashes":  fuzzy_json or "[]",
        "ip":            ip or "",
        "mail_from":     (mail_from or "").lower(),
        "mime_from":     mime["mime_from"],
        "rcpt_to":       "unknown" if not rcpt_list else ", ".join(r.lower() for r in rcpt_list),
        "mime_to":       mime["mime_to"],
        "mail_stored":   mail_stored,
        "mail_location": mail_location,
        "size":          size,
        "headers":       mime["headers_raw"],
        "message_id":    mime["message_id"],
    }

    data = _trim_fields(data)
    return _insert(data, rcpt_list)


# ─── Dipendenza: ACL + autenticazione ─────────────────────────────────────────

async def _auth(
    request: Request,
    credentials: HTTPBasicCredentials = Depends(_security),
) -> None:
    client_ip = request.client.host if request.client else ""
    if client_ip not in API_ACL:
        log.warning(f"Richiesta da {client_ip} rifiutata (non in RSPAMD_API_ACL)")
        raise HTTPException(status_code=403, detail="Forbidden")

    ok = (
        secrets.compare_digest(credentials.username.encode(), API_USER.encode())
        and secrets.compare_digest(credentials.password.encode(), API_PASS.encode())
    )
    if not ok:
        raise HTTPException(
            status_code=401,
            detail="Unauthorized",
            headers={"WWW-Authenticate": 'Basic realm="rqwatch-api"'},
        )


# ─── Endpoint: multipart/form-data (formatter = "multipart") ─────────────────

@app.post("/api/metadata_importer_multipart", response_class=PlainTextResponse)
async def metadata_importer_multipart(
    request: Request,
    metadata: str = Form(...),
    message: UploadFile = File(...),
    _: None = Depends(_auth),
) -> str:
    try:
        meta: dict = json.loads(metadata)
    except json.JSONDecodeError as exc:
        raise HTTPException(status_code=400, detail=f"metadata JSON non valido: {exc}")

    raw_email = await message.read()
    if not raw_email:
        raise HTTPException(status_code=400, detail="File message vuoto")

    qid = str(meta.get("qid") or "unknown")
    if not re.match(r"^[a-zA-Z0-9]+$", qid):
        qid = "unknown"

    score = float(meta.get("score") or 0.0)
    action = str(meta.get("action") or "")
    server = request.query_params.get("server", SERVER_ALIAS)

    if not qid and not score and not action:
        raise HTTPException(status_code=400, detail="qid, score e action mancanti")

    rcpt = meta.get("rcpt", [])
    if isinstance(rcpt, str) and rcpt not in ("", "unknown"):
        rcpt = [rcpt]
    elif not isinstance(rcpt, list):
        rcpt = []
    rcpt = [r.lower().strip() for r in rcpt if isinstance(r, str) and r.strip()]

    fuzzy = meta.get("fuzzy")
    if isinstance(fuzzy, list):
        fuzzy_json = json.dumps(fuzzy, ensure_ascii=False)
    elif fuzzy in (None, "", "unknown"):
        fuzzy_json = "[]"
    else:
        fuzzy_json = str(fuzzy)

    symbols = meta.get("symbols", {})
    if isinstance(symbols, (dict, list)):
        symbols_json = json.dumps(symbols, ensure_ascii=False)
    else:
        symbols_json = str(symbols) if symbols else "[]"

    try:
        db_id = _process(
            qid=qid,
            server=server,
            subject=str(meta.get("subject") or ""),
            score=score,
            action=action,
            symbols_json=symbols_json,
            fuzzy_json=fuzzy_json,
            ip=str(meta.get("ip") or ""),
            mail_from=str(meta.get("from") or ""),
            rcpt_list=rcpt,
            size=int(meta.get("size") or 0),
            raw_email=raw_email,
        )
    except Exception as exc:
        log.error(f"{qid} errore DB: {exc}")
        raise HTTPException(status_code=500, detail="Errore database")

    log.info(f"{qid} score:{score:.2f} action:'{action}' salvato [id:{db_id}]")
    return "Message saved"


# ─── Endpoint: raw body + X-Rspamd-* headers (formatter = "default") ─────────

@app.post("/api/metadata_importer", response_class=PlainTextResponse)
async def metadata_importer(
    request: Request,
    _: None = Depends(_auth),
) -> str:
    raw_email = await request.body()
    if not raw_email:
        raise HTTPException(status_code=400, detail="Body vuoto")

    h = request.headers
    qid = h.get("x-rspamd-qid", "unknown")
    if qid and not re.match(r"^[a-zA-Z0-9]+$", qid):
        qid = "unknown"

    action = h.get("x-rspamd-action", "")
    server = request.query_params.get("server", SERVER_ALIAS)

    try:
        score = float(h.get("x-rspamd-score") or "0")
    except ValueError:
        score = 0.0

    try:
        size = int(h.get("x-rspamd-size") or "0")
    except ValueError:
        size = 0

    symbols_raw = h.get("x-rspamd-symbols", "[]")
    fuzzy_raw = h.get("x-rspamd-fuzzy", "[]")
    fuzzy_json = "[]" if fuzzy_raw in ("unknown", "", None) else fuzzy_raw

    rcpt_raw = h.get("x-rspamd-rcpt", "[]")
    try:
        rcpt = json.loads(rcpt_raw) if rcpt_raw not in ("", "unknown") else []
        if not isinstance(rcpt, list):
            rcpt = [str(rcpt)] if rcpt else []
    except Exception:
        rcpt = []
    rcpt = [r.lower().strip() for r in rcpt if isinstance(r, str) and r.strip()]

    try:
        db_id = _process(
            qid=qid,
            server=server,
            subject=h.get("x-rspamd-subject", ""),
            score=score,
            action=action,
            symbols_json=symbols_raw or "[]",
            fuzzy_json=fuzzy_json,
            ip=h.get("x-rspamd-ip", ""),
            mail_from=h.get("x-rspamd-from", ""),
            rcpt_list=rcpt,
            size=size,
            raw_email=raw_email,
        )
    except Exception as exc:
        log.error(f"{qid} errore DB: {exc}")
        raise HTTPException(status_code=500, detail="Errore database")

    log.info(f"{qid} score:{score:.2f} action:'{action}' salvato [id:{db_id}]")
    return "Message saved"


# ─── Entry point ──────────────────────────────────────────────────────────────

if __name__ == "__main__":
    import uvicorn

    uvicorn.run(
        "main:app",
        host=os.getenv("LISTEN_HOST", "127.0.0.1"),
        port=int(os.getenv("LISTEN_PORT", "8080")),
        reload=False,
    )