from __future__ import annotations

import copy
import tempfile
import unittest
from pathlib import Path
from types import SimpleNamespace
from unittest.mock import patch

from govoplan_campaign.backend.attachments.resolver import (
    MessageAttachmentStatus,
    resolve_entry_attachments,
)
from govoplan_campaign.backend.campaign.models import CampaignConfig
from govoplan_campaign.backend.path_security import (
    CampaignPathSecurityError,
    assert_server_safe_campaign_paths,
)
from govoplan_campaign.backend.persistence.campaigns import (
    create_campaign_version_from_json,
    load_version_config,
)
from govoplan_campaign.backend.sending.execution import ExecutionSnapshotError, ensure_execution_snapshot


def _campaign_json() -> dict[str, object]:
    return {
        "version": "1.0",
        "campaign": {"id": "path-security", "name": "Path security", "mode": "test"},
        "template": {"subject": "Subject", "text": "Body"},
        "attachments": {"base_path": ".", "base_paths": [], "global": []},
        "entries": {"inline": []},
    }


class ServerCampaignPathSecurityTests(unittest.TestCase):
    def test_api_import_rejects_absolute_source_filename_before_persistence(self) -> None:
        with patch(
            "govoplan_campaign.backend.persistence.campaigns.files_integration",
            return_value=SimpleNamespace(available=False),
        ):
            with self.assertRaisesRegex(CampaignPathSecurityError, "source_filename"):
                create_campaign_version_from_json(
                    None,  # type: ignore[arg-type]
                    tenant_id="tenant-1",
                    user_id="user-1",
                    raw_json=_campaign_json(),
                    source_filename="/etc/passwd",
                )

    def test_template_parent_traversal_is_rejected(self) -> None:
        raw = _campaign_json()
        raw["template"] = {"source": {"text_path": "../../etc/passwd"}}

        with self.assertRaisesRegex(CampaignPathSecurityError, "template source paths"):
            assert_server_safe_campaign_paths(raw, managed_files_available=True)

    def test_existing_unsafe_version_cannot_reuse_a_delivery_snapshot(self) -> None:
        raw = _campaign_json()
        raw["template"] = {"source": {"text_path": "/etc/passwd"}}
        version = SimpleNamespace(raw_json=raw, execution_snapshot={})

        with patch(
            "govoplan_campaign.backend.sending.execution.files_integration",
            return_value=SimpleNamespace(available=True),
        ):
            with self.assertRaisesRegex(ExecutionSnapshotError, "template source paths"):
                ensure_execution_snapshot(None, version)  # type: ignore[arg-type]

    def test_existing_unsafe_version_is_rejected_before_config_loading(self) -> None:
        raw = _campaign_json()
        raw["template"] = {"source": {"text_path": "/etc/passwd"}}
        version = SimpleNamespace(campaign_id="campaign-1", raw_json=raw)
        campaign = SimpleNamespace(id="campaign-1", tenant_id="tenant-1")
        session = SimpleNamespace(get=lambda _model, key: version if key == "version-1" else campaign)

        with (
            patch(
                "govoplan_campaign.backend.persistence.campaigns.files_integration",
                return_value=SimpleNamespace(available=True),
            ),
            patch("govoplan_campaign.backend.persistence.campaigns._write_campaign_snapshot") as write_snapshot,
            patch("govoplan_campaign.backend.persistence.campaigns.load_campaign_config_from_json") as load_config,
        ):
            with self.assertRaisesRegex(CampaignPathSecurityError, "template source paths"):
                load_version_config(session, "version-1")  # type: ignore[arg-type]

        write_snapshot.assert_not_called()
        load_config.assert_not_called()

    def test_source_base_symlink_escape_is_rejected(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            outside = root / "outside"
            outside.mkdir()
            (outside / "secret.txt").write_text("secret", encoding="utf-8")
            trusted = root / "trusted"
            trusted.mkdir()
            (trusted / "escape").symlink_to(outside, target_is_directory=True)

            raw = _campaign_json()
            raw["template"] = {"source": {"text_path": "escape/secret.txt"}}
            with self.assertRaisesRegex(CampaignPathSecurityError, "source_base_path"):
                assert_server_safe_campaign_paths(
                    raw,
                    source_base_path=str(trusted),
                    managed_files_available=True,
                )

    def test_attachment_absolute_path_and_parent_filter_are_rejected(self) -> None:
        raw = _managed_attachment_campaign()
        attachments = raw["attachments"]
        assert isinstance(attachments, dict)
        base_paths = attachments["base_paths"]
        assert isinstance(base_paths, list)
        assert isinstance(base_paths[0], dict)
        base_paths[0]["path"] = "/etc"
        with self.assertRaisesRegex(CampaignPathSecurityError, "relative managed-file path"):
            assert_server_safe_campaign_paths(raw, managed_files_available=True)

        raw = _managed_attachment_campaign()
        attachments = raw["attachments"]
        assert isinstance(attachments, dict)
        rules = attachments["global"]
        assert isinstance(rules, list)
        assert isinstance(rules[0], dict)
        rules[0]["file_filter"] = "../../etc/passwd"
        with self.assertRaisesRegex(CampaignPathSecurityError, "parent-directory traversal"):
            assert_server_safe_campaign_paths(raw, managed_files_available=True)

    def test_local_attachment_rule_is_rejected_in_server_mode(self) -> None:
        raw = _managed_attachment_campaign()
        attachments = raw["attachments"]
        assert isinstance(attachments, dict)
        base_paths = attachments["base_paths"]
        assert isinstance(base_paths, list)
        assert isinstance(base_paths[0], dict)
        base_paths[0].pop("source")

        with self.assertRaisesRegex(CampaignPathSecurityError, "backed by managed Files"):
            assert_server_safe_campaign_paths(raw, managed_files_available=True)

    def test_default_attachment_rule_must_use_managed_files(self) -> None:
        raw = _managed_attachment_campaign()
        attachments = raw["attachments"]
        assert isinstance(attachments, dict)
        attachments["global"] = []
        entries = raw["entries"]
        assert isinstance(entries, dict)
        entries["defaults"] = {
            "attachments": [{"base_dir": "invoices", "file_filter": "invoice.pdf"}]
        }

        with self.assertRaisesRegex(CampaignPathSecurityError, "entries.defaults.attachments"):
            assert_server_safe_campaign_paths(raw, managed_files_available=True)

    def test_normal_managed_file_campaign_is_allowed(self) -> None:
        raw = _managed_attachment_campaign()
        assert_server_safe_campaign_paths(raw, managed_files_available=True)

        copied = copy.deepcopy(raw)
        attachments = copied["attachments"]
        assert isinstance(attachments, dict)
        rules = attachments["global"]
        assert isinstance(rules, list)
        assert isinstance(rules[0], dict)
        rules[0]["file_filter"] = "**/{{local:invoice_number}}-report.XLSX"
        assert_server_safe_campaign_paths(copied, managed_files_available=True)

    def test_rendered_managed_filter_cannot_traverse_materialized_source(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            managed_root = root / "managed"
            managed_root.mkdir()
            (root / "secret.pdf").write_bytes(b"server secret")

            config = _runtime_managed_attachment_config(
                managed_root,
                file_filter="{{local:file_name}}",
                file_name="../secret.pdf",
            )
            resolution = resolve_entry_attachments(
                config=config,
                campaign_file=root / "campaign.json",
                entry=config.entries.inline[0],  # type: ignore[index]
                entry_index=1,
            )

        self.assertEqual(resolution.status, MessageAttachmentStatus.BLOCKED)
        self.assertEqual(resolution.attachments[0].matches, [])
        self.assertIn(
            "unsafe_managed_attachment_path",
            {issue.code for issue in resolution.issues},
        )

    def test_managed_match_symlink_cannot_escape_materialized_source(self) -> None:
        with tempfile.TemporaryDirectory() as temp_dir:
            root = Path(temp_dir)
            managed_root = root / "managed"
            managed_root.mkdir()
            outside = root / "secret.pdf"
            outside.write_bytes(b"server secret")
            (managed_root / "escape.pdf").symlink_to(outside)

            config = _runtime_managed_attachment_config(
                managed_root,
                file_filter="escape.pdf",
                file_name="unused.pdf",
            )
            resolution = resolve_entry_attachments(
                config=config,
                campaign_file=root / "campaign.json",
                entry=config.entries.inline[0],  # type: ignore[index]
                entry_index=1,
            )

        self.assertEqual(resolution.status, MessageAttachmentStatus.BLOCKED)
        self.assertEqual(resolution.attachments[0].matches, [])
        self.assertIn(
            "managed_attachment_path_escape",
            {issue.code for issue in resolution.issues},
        )


def _managed_attachment_campaign() -> dict[str, object]:
    raw = _campaign_json()
    raw["attachments"] = {
        "base_path": "invoices",
        "base_paths": [
            {
                "id": "personal-invoices",
                "name": "My invoices",
                "source": "managed:user:user-1",
                "path": "invoices",
            }
        ],
        "global": [
            {
                "base_path_id": "personal-invoices",
                "base_dir": "invoices",
                "file_filter": "invoice.pdf",
            }
        ],
    }
    return raw


def _runtime_managed_attachment_config(
    managed_root: Path,
    *,
    file_filter: str,
    file_name: str,
) -> CampaignConfig:
    raw = _managed_attachment_campaign()
    raw["fields"] = [{"name": "file_name", "type": "string", "required": True}]
    attachments = raw["attachments"]
    assert isinstance(attachments, dict)
    base_paths = attachments["base_paths"]
    assert isinstance(base_paths, list)
    assert isinstance(base_paths[0], dict)
    base_paths[0]["path"] = str(managed_root)
    rules = attachments["global"]
    assert isinstance(rules, list)
    assert isinstance(rules[0], dict)
    rules[0]["file_filter"] = file_filter
    entries = raw["entries"]
    assert isinstance(entries, dict)
    entries["inline"] = [
        {
            "id": "recipient-1",
            "to": [{"email": "recipient@example.test", "type": "to"}],
            "fields": {"file_name": file_name},
        }
    ]
    return CampaignConfig.model_validate(raw)


if __name__ == "__main__":
    unittest.main()