volumen/tests/test_security.py

"""Security-focused tests: signature validation, sanitisation, cookies, headers."""

from __future__ import annotations

import io

from fastapi.testclient import TestClient

from volumen.markdown import render
from volumen.password import (
    BLOCK_R,
    COST_N,
    KEY_LENGTH,
    PARALLEL_P,
    hash_password,
    needs_rehash,
    verify_password,
)
from volumen.store import (
    ALLOWED_IMAGE_EXTENSIONS,
    detect_image_extension,
    validate_image_signature,
)

# ---------------------------------------------------------------------------
# Image signature validation
# ---------------------------------------------------------------------------


class TestImageSignature:
    def test_webp_magic_accepted(self) -> None:
        # Minimal RIFF/WEBP header (12 bytes).
        good = b"RIFF" + b"\x10\x00\x00\x00" + b"WEBP" + b"VP8 "
        assert validate_image_signature(good, ".webp") is True

    def test_webp_wrong_magic_rejected(self) -> None:
        # RIFF but not WEBP at offset 8.
        bad = b"RIFF" + b"\x10\x00\x00\x00" + b"WAVE" + b"VP8 "
        assert validate_image_signature(bad, ".webp") is False

    def test_webp_too_short_rejected(self) -> None:
        assert validate_image_signature(b"RIFF", ".webp") is False

    def test_avif_magic_accepted(self) -> None:
        # ftyp box with size=0x18, brand=avif.
        avif = b"\x00\x00\x00\x18ftyp" + b"avif" + b"\x00\x00\x00\x00" + b"avif" + b"mif1"
        assert validate_image_signature(avif, ".avif") is True

    def test_avif_alt_brand_accepted(self) -> None:
        avif = b"\x00\x00\x00\x18ftyp" + b"avis" + b"\x00\x00\x00\x00" + b"avis" + b"mif1"
        assert validate_image_signature(avif, ".avif") is True

    def test_avif_wrong_brand_rejected(self) -> None:
        avif = b"\x00\x00\x00\x18ftyp" + b"mp42" + b"\x00\x00\x00\x00" + b"mp42" + b"isom"
        assert validate_image_signature(avif, ".avif") is False

    def test_avif_too_short_rejected(self) -> None:
        assert validate_image_signature(b"\x00\x00\x00\x18ftyp", ".avif") is False

    def test_unknown_extension_rejected(self) -> None:
        assert validate_image_signature(b"anything", ".png") is False

    def test_detect_extension_webp(self) -> None:
        data = b"RIFF" + b"\x00\x00\x00\x00" + b"WEBP"
        assert detect_image_extension(data) == ".webp"

    def test_detect_extension_avif(self) -> None:
        data = b"\x00\x00\x00\x18ftypavif" + b"\x00" * 8
        assert detect_image_extension(data) == ".avif"

    def test_detect_extension_unknown(self) -> None:
        assert detect_image_extension(b"random") is None

    def test_allowed_extensions_match_policy(self) -> None:
        assert ALLOWED_IMAGE_EXTENSIONS == (".webp", ".avif")


# ---------------------------------------------------------------------------
# Markdown XSS sanitisation
# ---------------------------------------------------------------------------


class TestMarkdownSanitisation:
    def test_script_tag_stripped(self) -> None:
        rendered = render("Hello <script>alert(1)</script>world")
        assert "<script" not in rendered.lower()
        assert "alert(1)" not in rendered

    def test_inline_event_handler_stripped(self) -> None:
        rendered = render('<img src="x" onerror="alert(1)">')
        assert "onerror" not in rendered.lower()
        assert "alert" not in rendered.lower()

    def test_javascript_url_sanitised(self) -> None:
        rendered = render("[click me](javascript:alert(1))")
        assert "javascript:" not in rendered.lower()

    def test_data_image_url_kept(self) -> None:
        rendered = render("![alt](data:image/webp;base64,UklGRg==)")
        assert "<img" in rendered

    def test_figure_wrapping_still_works(self) -> None:
        rendered = render('![alt](url "caption")')
        assert "<figure>" in rendered
        assert "caption" in rendered


# ---------------------------------------------------------------------------
# Password hashing policy
# ---------------------------------------------------------------------------


class TestPasswordPolicy:
    def test_strong_hash_passes(self) -> None:
        encoded = hash_password("a-strong-password")
        assert verify_password("a-strong-password", encoded) is True

    def test_weak_params_rejected(self) -> None:
        encoded = f"scrypt$8$1$1${'A' * 24}${'B' * 32}"
        assert verify_password("anything", encoded) is False

    def test_needs_rehash_for_strong_hash(self) -> None:
        encoded = hash_password("a-strong-password")
        assert needs_rehash(encoded) is False

    def test_needs_rehash_for_weak_hash(self) -> None:
        encoded = f"scrypt$8$1$1${'A' * 24}${'B' * 32}"
        assert needs_rehash(encoded) is True

    def test_needs_rehash_for_malformed(self) -> None:
        assert needs_rehash("not-a-hash") is True

    def test_module_constants_match_policy(self) -> None:
        # Policy floor equals current implementation.
        assert COST_N >= 16384
        assert BLOCK_R >= 8
        assert PARALLEL_P >= 1
        assert KEY_LENGTH >= 32


# ---------------------------------------------------------------------------
# HTTP security headers and CSP
# ---------------------------------------------------------------------------


class TestSecurityHeaders:
    def test_global_security_headers_present(self, client: TestClient) -> None:
        response = client.get("/api/volumen/site")
        assert response.headers.get("x-content-type-options") == "nosniff"
        assert response.headers.get("x-frame-options") == "DENY"
        assert response.headers.get("referrer-policy") == "strict-origin-when-cross-origin"
        assert "permissions-policy" in response.headers

    def test_csp_on_public_routes(self, client: TestClient) -> None:
        response = client.get("/api/volumen/site")
        csp = response.headers.get("content-security-policy", "")
        assert "default-src 'self'" in csp
        # Public routes have script-src but without nonce (nonce is admin-only)
        assert "script-src 'self'" in csp
        assert "nonce-" not in csp

    def test_admin_route_has_csp_with_nonce(self, admin_client: TestClient) -> None:
        response = admin_client.get("/admin/login")
        csp = response.headers.get("content-security-policy", "")
        assert "default-src 'self'" in csp
        assert "'unsafe-inline'" not in csp
        assert "nonce-" in csp

    def test_admin_csp_has_no_duplicate_directives(self, admin_client: TestClient) -> None:
        # Regression: nonce-augmented script-src / style-src used to be
        # appended after the base ones, so the same directive name appeared
        # twice. Browsers use the FIRST occurrence and silently drop the
        # nonce, breaking all admin JS. Verify each directive is unique.
        response = admin_client.get("/admin/login")
        csp = response.headers.get("content-security-policy", "")
        directives = [d.strip().split(maxsplit=1)[0] for d in csp.split(";")]
        assert len(directives) == len(set(directives)), (
            f"Duplicate CSP directives: {[d for d in directives if directives.count(d) > 1]}"
        )
        # And specifically: only one script-src, only one style-src.
        assert directives.count("script-src") == 1
        assert directives.count("style-src") == 1

    def test_admin_csp_nonce_appears_in_rendered_html(self, admin_client: TestClient) -> None:
        # Regression: nonce used to be set AFTER call_next, so templates
        # rendered with an empty csp_nonce and the matching <script nonce="...">
        # tag in the response never carried the nonce that CSP demanded.
        # Browsers then blocked every inline script. Verify the nonce in
        # the response header matches the nonce on the <script> tags.
        import re

        response = admin_client.get("/admin/login")
        csp = response.headers.get("content-security-policy", "")
        match = re.search(r"nonce-([A-Za-z0-9_-]+)", csp)
        assert match is not None, "No nonce in CSP"
        nonce = match.group(1)
        # At least one inline <script> must carry this exact nonce.
        script_tags = re.findall(r"<script[^>]*>", response.text)
        assert script_tags, "No <script> tags rendered"
        assert any(f'nonce="{nonce}"' in tag for tag in script_tags), (
            "Rendered <script> tags do not carry the CSP nonce:\n" + "\n".join(script_tags)
        )

    def test_admin_pages_have_no_blocking_inline_styles(self, admin_client: TestClient) -> None:
        # Regression: with `style-src 'self' 'nonce-...'`, any HTML element
        # carrying `style="..."` is blocked by the browser — including
        # style attributes that do *not* start with `--` (CSS custom
        # properties, which are explicitly allowed by CSP3). Walk every
        # admin page and assert no offending attribute survives.
        import re

        def _offending(body: str) -> list[str]:
            results: list[str] = []
            for match in re.finditer(r"\sstyle=\"([^\"]*)\"", body):
                value = match.group(1).strip()
                # `style="--foo: ..."` is a CSS custom-property declaration
                # and is allowed by CSP3 without `unsafe-inline`. Anything
                # else is a style application and must move into a class.
                if not value.startswith("--"):
                    results.append(match.group(0))
            return results

        for path in ("/admin/login", "/admin/", "/admin/settings"):
            response = admin_client.get(path)
            assert response.status_code == 200, f"{path} returned {response.status_code}"
            offending = _offending(response.text)
            assert not offending, (
                f"{path} renders {len(offending)} non-custom-property inline "
                f"style attribute(s) that strict CSP will block:\n" + "\n".join(offending)
            )

        # Also check the post form (requires login) — it had several
        # one-off utility styles that should now be classes.
        from tests.test_admin import _login

        _login(admin_client)
        response = admin_client.get("/admin/posts/new")
        assert response.status_code == 200
        offending = _offending(response.text)
        assert not offending, (
            "/admin/posts/new renders inline style attributes that strict "
            "CSP will block:\n" + "\n".join(offending)
        )

    def test_favicon_ico_serves_svg(self, client: TestClient) -> None:
        # Regression: browsers auto-request /favicon.ico even when the page
        # declares an explicit <link rel="icon">, and the previous behaviour
        # was a 404. Serve the packaged icon SVG so the console is clean
        # and the tab gets a proper icon.
        response = client.get("/favicon.ico")
        assert response.status_code == 200
        assert response.headers["content-type"].startswith("image/svg+xml")
        body = response.content
        assert body.startswith(b"<") and b"</svg>" in body
        # And the admin login page must declare the icon explicitly.
        login = client.get("/admin/login")
        assert login.status_code == 200
        assert 'rel="icon"' in login.text
        assert 'href="/favicon.ico"' in login.text

    def test_admin_layout_renders_version(self, admin_client: TestClient) -> None:
        # Regression: `version` was not in admin_context, so the sidebar
        # footer rendered as just "v" with no number behind it.
        # The sidebar footer only renders on the authenticated layout,
        # so we log in first.
        from tests.test_admin import _login
        from volumen.version import VERSION

        _login(admin_client)
        response = admin_client.get("/admin/")
        assert response.status_code == 200
        assert "sidebar-version" in response.text
        # The version label is "v{version}" inside a div.sidebar-version.
        # Refuse the bare "v</div>" rendering that this bug produced.
        assert "v" + VERSION in response.text, (
            f"Expected the sidebar version label to include {VERSION}; "
            f"admin_context is not propagating the package version"
        )

    def test_https_only_cookie_unset_when_untrusted(self, client: TestClient) -> None:
        # Default trust_proxy=false — X-Forwarded-Proto is ignored.
        response = client.get("/admin/login")
        # No cookie issued, but the middleware must mark session_secure=False.
        assert response.status_code == 200

    def test_https_only_cookie_with_trusted_proxy(self, admin_config_dir) -> None:
        # Build a fresh app with trust_proxy=True and confirm it boots.
        from volumen.app import create_app
        from volumen.config import Config
        from volumen.store import Store

        config = Config.load(path=str(admin_config_dir / "config.toml"))
        config._data["server"]["trust_proxy"] = True  # type: ignore[index]
        store = Store(config.content_dir)
        app = create_app(config, store)
        client = TestClient(app)
        resp = client.get("/admin/login")
        assert resp.status_code == 200

    def test_cookie_secure_false_without_proxy_or_flag(self, admin_client: TestClient) -> None:
        # When trust_proxy=false and cookie_secure=false, the SessionMiddleware
        # is configured with https_only=False.
        resp = admin_client.get("/admin/login")
        assert resp.status_code == 200


# ---------------------------------------------------------------------------
# Config validation
# ---------------------------------------------------------------------------


class TestConfigValidation:
    def test_default_config_is_valid(self, config) -> None:
        # Should not raise.
        config.validate()

    def test_invalid_port_rejected(self, tmp_path) -> None:
        from volumen.config import Config

        (tmp_path / "posts").mkdir()
        cfg = Config.load(
            path="/nonexistent.toml",
            overrides={
                "users_file": str(tmp_path / "users.toml"),
                "content": str(tmp_path / "posts"),
            },
        )
        cfg._data["server"]["port"] = 999999  # type: ignore[index]
        try:
            cfg.validate()
        except Exception as exc:
            assert "port" in str(exc).lower()
        else:
            raise AssertionError("expected ConfigError")

    def test_invalid_env_rejected(self, tmp_path) -> None:
        from volumen.config import Config, ConfigError

        cfg = Config.load(
            path="/nonexistent.toml",
            overrides={"users_file": str(tmp_path / "users.toml")},
        )
        cfg._data["server"]["env"] = "staging"  # type: ignore[index]
        try:
            cfg.validate()
        except ConfigError as exc:
            assert "env" in str(exc).lower()
        else:
            raise AssertionError("expected ConfigError")

    def test_invalid_base_url_rejected(self, tmp_path) -> None:
        from volumen.config import Config, ConfigError

        cfg = Config.load(
            path="/nonexistent.toml",
            overrides={"users_file": str(tmp_path / "users.toml")},
        )
        cfg._data["site"]["base_url"] = "not-a-url"  # type: ignore[index]
        try:
            cfg.validate()
        except ConfigError as exc:
            assert "base_url" in str(exc).lower()
        else:
            raise AssertionError("expected ConfigError")

    def test_production_requires_session_key(self, tmp_path) -> None:
        from volumen.config import Config

        (tmp_path / "posts").mkdir()
        cfg = Config.load(
            path="/nonexistent.toml",
            overrides={
                "users_file": str(tmp_path / "users.toml"),
                "content": str(tmp_path / "posts"),
            },
        )
        cfg._data["server"]["env"] = "production"  # type: ignore[index]
        cfg._data["admin"]["session_key"] = ""  # type: ignore[index]
        # Validation succeeds (file is allowed) but create_app will refuse.
        cfg.validate()


# ---------------------------------------------------------------------------
# Session secret validation
# ---------------------------------------------------------------------------


class TestSessionSecret:
    def test_production_without_session_key_fails(self, config) -> None:
        from volumen.app import create_app
        from volumen.store import Store

        config._data["server"]["env"] = "production"  # type: ignore[index]
        config._data["admin"]["session_key"] = ""  # type: ignore[index]
        store = Store(config.content_dir)
        try:
            create_app(config, store)
        except RuntimeError as exc:
            assert "session_key" in str(exc)
        else:
            raise AssertionError("expected RuntimeError")

    def test_production_with_short_key_fails(self, config) -> None:
        from volumen.app import create_app
        from volumen.store import Store

        config._data["server"]["env"] = "production"  # type: ignore[index]
        config._data["admin"]["session_key"] = "tooshort"  # type: ignore[index]
        store = Store(config.content_dir)
        try:
            create_app(config, store)
        except RuntimeError as exc:
            assert "64 bytes" in str(exc)
        else:
            raise AssertionError("expected RuntimeError")


# ---------------------------------------------------------------------------
# Password policy helpers
# ---------------------------------------------------------------------------


class TestPasswordPolicyHelpers:
    def test_password_error_min_length(self) -> None:
        from volumen.auth import password_error

        assert password_error("short", min_length=10) is not None

    def test_password_error_max_length(self) -> None:
        from volumen.auth import password_error

        assert password_error("x" * 2000, min_length=10, max_length=1024) is not None

    def test_password_error_empty(self) -> None:
        from volumen.auth import password_error

        assert password_error("", min_length=10) is not None

    def test_password_error_common_rejected(self) -> None:
        from volumen.auth import password_error

        assert password_error("password", min_length=10) is not None

    def test_password_error_accepts_strong(self) -> None:
        from volumen.auth import password_error

        assert password_error("volumen-strong-pw", min_length=10) is None


# ---------------------------------------------------------------------------
# Store hardening
# ---------------------------------------------------------------------------


class _Upload:
    """Fake UploadFile for testing validate_upload."""

    content_type: str
    filename: str
    file: io.BytesIO

    def __init__(
        self, data: bytes, content_type: str = "image/webp", filename: str = "x.webp"
    ) -> None:
        self._data = data
        self.content_type = content_type
        self.filename = filename
        self.file = io.BytesIO(data)
        self._pos = 0

    async def read(self, size: int = -1) -> bytes:
        if size == -1:
            result = self._data[self._pos :]
            self._pos = len(self._data)
            return result
        result = self._data[self._pos : self._pos + size]
        self._pos += size
        return result


class TestStoreHardening:
    def test_validate_upload_accepts_webp(self, tmp_content_dir) -> None:
        # Build a fake UploadFile-shaped object.
        good_webp = b"RIFF" + b"\x10\x00\x00\x00" + b"WEBP" + b"\x00" * 100

        import asyncio

        from volumen.upload_validation import validate_upload

        async def go() -> tuple[bytes, str]:
            return await validate_upload(_Upload(good_webp), max_bytes=10_000)

        data, ext = asyncio.run(go())
        assert ext == ".webp"
        assert data == good_webp

    def test_validate_upload_rejects_mime_mismatch(self, tmp_content_dir) -> None:
        import asyncio

        from volumen.upload_validation import validate_upload

        async def go() -> None:
            try:
                await validate_upload(
                    _Upload(b"nope", content_type="image/png", filename="x.png"),
                    max_bytes=10_000,
                )
            except Exception as exc:
                # The detail is JSON-encoded text.
                msg = str(exc)
                assert "unsupported_format" in msg or "signature" in msg.lower()
            else:
                raise AssertionError("expected HTTPException")

        asyncio.run(go())

    def test_store_upload_uses_signature_extension(self, tmp_content_dir) -> None:
        from volumen.store import Store

        store = Store(str(tmp_content_dir))
        avif_bytes = b"\x00\x00\x00\x18ftyp" + b"avif" + b"\x00" * 100
        url = store.store_upload("evil.exe", None, bytes_data=avif_bytes)
        assert url.endswith(".avif")
        assert "evil" not in url

    def test_store_upload_resolves_whitelisted_extension(self, tmp_content_dir) -> None:
        from volumen.store import Store

        store = Store(str(tmp_content_dir))
        webp_bytes = b"RIFF" + b"\x10\x00\x00\x00" + b"WEBP" + b"\x00" * 100
        # File name has an unrelated extension; signature wins.
        url = store.store_upload("cover.png", None, bytes_data=webp_bytes)
        assert url.endswith(".webp")