testy

test/test_security.py

@@ -0,0 +1,598 @@
+from __future__ import annotations
+
+import hashlib
+import hmac
+import json
+
+import pytest
+from fastapi.testclient import TestClient
+
+import app.main as main
+import app.routes as routes
+import app.security as security
+
+
+WEBHOOK_SECRET = "w" * 64
+SYNC_API_KEY = "s" * 64
+SEARCH_API_KEY = "a" * 64
+
+EXPECTED_REPOSITORY = (
+    "KEMT/zpwiki"
+)
+
+
+def sign(
+    body: bytes,
+) -> str:
+    return hmac.new(
+        WEBHOOK_SECRET.encode(
+            "utf-8"
+        ),
+        body,
+        hashlib.sha256,
+    ).hexdigest()
+
+
+@pytest.fixture
+def client(
+    security_environment,
+) -> TestClient:
+    with TestClient(
+        main.app
+    ) as test_client:
+        yield test_client
+
+
+def test_security_configuration_accepts_valid_configuration(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv(
+        "WEBHOOK_SECRET",
+        WEBHOOK_SECRET,
+    )
+
+    monkeypatch.setenv(
+        "SYNC_API_KEY",
+        SYNC_API_KEY,
+    )
+
+    monkeypatch.setenv(
+        "SEARCH_API_KEY",
+        SEARCH_API_KEY,
+    )
+
+    monkeypatch.setenv(
+        "EXPECTED_GITEA_REPOSITORY",
+        EXPECTED_REPOSITORY,
+    )
+
+    monkeypatch.setenv(
+        "WEBHOOK_PULL_GIT",
+        "false",
+    )
+
+    security.validate_security_configuration()
+
+
+def test_security_configuration_rejects_shared_secrets(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    shared = "x" * 64
+
+    monkeypatch.setenv(
+        "WEBHOOK_SECRET",
+        shared,
+    )
+
+    monkeypatch.setenv(
+        "SYNC_API_KEY",
+        shared,
+    )
+
+    monkeypatch.setenv(
+        "SEARCH_API_KEY",
+        SEARCH_API_KEY,
+    )
+
+    monkeypatch.setenv(
+        "EXPECTED_GITEA_REPOSITORY",
+        EXPECTED_REPOSITORY,
+    )
+
+    with pytest.raises(
+        RuntimeError,
+        match="navzájom rozdielne",
+    ):
+        security.validate_security_configuration()
+
+
+@pytest.mark.parametrize(
+    "value",
+    [
+        "zpwiki",
+        "/zpwiki",
+        "KEMT/",
+        "KEMT/zpwiki/extra",
+        "KEMT /zpwiki",
+        "KEMT/zp wiki",
+    ],
+)
+def test_expected_repository_rejects_invalid_format(
+    monkeypatch: pytest.MonkeyPatch,
+    value: str,
+) -> None:
+    monkeypatch.setenv(
+        "EXPECTED_GITEA_REPOSITORY",
+        value,
+    )
+
+    with pytest.raises(
+        RuntimeError,
+        match="vlastník/repozitár",
+    ):
+        security.expected_gitea_repository()
+
+
+@pytest.mark.parametrize(
+    (
+        "value",
+        "expected",
+    ),
+    [
+        ("true", True),
+        ("TRUE", True),
+        ("1", True),
+        ("yes", True),
+        ("on", True),
+        ("false", False),
+        ("FALSE", False),
+        ("0", False),
+        ("no", False),
+        ("off", False),
+    ],
+)
+def test_webhook_pull_git_boolean_values(
+    monkeypatch: pytest.MonkeyPatch,
+    value: str,
+    expected: bool,
+) -> None:
+    monkeypatch.setenv(
+        "WEBHOOK_PULL_GIT",
+        value,
+    )
+
+    assert (
+        security.webhook_should_pull_git()
+        is expected
+    )
+
+
+def test_webhook_pull_git_rejects_invalid_value(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    monkeypatch.setenv(
+        "WEBHOOK_PULL_GIT",
+        "maybe",
+    )
+
+    with pytest.raises(
+        RuntimeError,
+        match="boolean",
+    ):
+        security.webhook_should_pull_git()
+
+
+def test_repository_payload_requires_full_name() -> None:
+    payload = {
+        "repository": {
+            "name": "zpwiki",
+        }
+    }
+
+    assert (
+        security.repository_name_from_payload(
+            payload
+        )
+        is None
+    )
+
+
+def test_repository_payload_accepts_full_name() -> None:
+    payload = {
+        "repository": {
+            "full_name": (
+                "KEMT/zpwiki"
+            ),
+        }
+    }
+
+    assert (
+        security.repository_name_from_payload(
+            payload
+        )
+        == "KEMT/zpwiki"
+    )
+
+
+def test_repository_comparison_is_case_insensitive() -> None:
+    assert (
+        security.same_repository(
+            "kemt/ZPWIKI",
+            "KEMT/zpwiki",
+        )
+        is True
+    )
+
+
+def test_gitea_signature_accepts_valid_digest() -> None:
+    body = b'{"test":true}'
+
+    signature = sign(
+        body
+    )
+
+    assert (
+        security.verify_gitea_signature(
+            body,
+            signature,
+            WEBHOOK_SECRET,
+        )
+        is True
+    )
+
+
+def test_gitea_signature_accepts_sha256_prefix() -> None:
+    body = b'{"test":true}'
+
+    signature = (
+        "sha256="
+        + sign(
+            body
+        )
+    )
+
+    assert (
+        security.verify_gitea_signature(
+            body,
+            signature,
+            WEBHOOK_SECRET,
+        )
+        is True
+    )
+
+
+def test_gitea_signature_rejects_modified_body() -> None:
+    original = b'{"test":true}'
+
+    modified = b'{"test":false}'
+
+    signature = sign(
+        original
+    )
+
+    assert (
+        security.verify_gitea_signature(
+            modified,
+            signature,
+            WEBHOOK_SECRET,
+        )
+        is False
+    )
+
+
+@pytest.mark.parametrize(
+    "signature",
+    [
+        None,
+        "",
+        "abc",
+        "x" * 64,
+        "0" * 63,
+        "0" * 65,
+    ],
+)
+def test_gitea_signature_rejects_invalid_signature(
+    signature: str | None,
+) -> None:
+    assert (
+        security.verify_gitea_signature(
+            b"body",
+            signature,
+            WEBHOOK_SECRET,
+        )
+        is False
+    )
+
+
+def test_sync_key_cannot_access_rag(
+    client: TestClient,
+) -> None:
+    response = client.post(
+        "/rag",
+        headers={
+            "X-API-Key": (
+                SYNC_API_KEY
+            ),
+        },
+        json={
+            "query": "Ján Holp",
+        },
+    )
+
+    assert (
+        response.status_code
+        == 401
+    )
+
+
+def test_search_key_cannot_access_sync(
+    client: TestClient,
+) -> None:
+    response = client.post(
+        "/sync",
+        headers={
+            "X-API-Key": (
+                SEARCH_API_KEY
+            ),
+        },
+        json={
+            "pull_git": False,
+        },
+    )
+
+    assert (
+        response.status_code
+        == 401
+    )
+
+
+def test_sync_key_cannot_be_used_as_rag_bearer(
+    client: TestClient,
+) -> None:
+    response = client.post(
+        "/rag",
+        headers={
+            "Authorization": (
+                f"Bearer {SYNC_API_KEY}"
+            ),
+        },
+        json={
+            "query": "Ján Holp",
+        },
+    )
+
+    assert (
+        response.status_code
+        == 401
+    )
+
+
+def test_health_does_not_expose_filesystem_paths(
+    client: TestClient,
+) -> None:
+    response = client.get(
+        "/health"
+    )
+
+    assert (
+        response.status_code
+        == 200
+    )
+
+    payload = response.json()
+
+    assert (
+        "database_path"
+        not in payload
+    )
+
+    assert (
+        "zpwiki_root"
+        not in payload
+    )
+
+    assert (
+        "/home/"
+        not in response.text
+    )
+
+
+def test_swagger_docs_are_disabled(
+    client: TestClient,
+) -> None:
+    assert (
+        client.get(
+            "/docs"
+        ).status_code
+        == 404
+    )
+
+    assert (
+        client.get(
+            "/redoc"
+        ).status_code
+        == 404
+    )
+
+
+def test_openapi_schema_remains_available(
+    client: TestClient,
+) -> None:
+    response = client.get(
+        "/openapi.json"
+    )
+
+    assert (
+        response.status_code
+        == 200
+    )
+
+    assert (
+        "/rag"
+        in response.json()[
+            "paths"
+        ]
+    )
+
+
+def test_security_response_headers(
+    client: TestClient,
+) -> None:
+    response = client.get(
+        "/health"
+    )
+
+    assert (
+        response.headers[
+            "cache-control"
+        ]
+        == "no-store"
+    )
+
+    assert (
+        response.headers[
+            "x-content-type-options"
+        ]
+        == "nosniff"
+    )
+
+    assert (
+        response.headers[
+            "referrer-policy"
+        ]
+        == "no-referrer"
+    )
+
+    assert (
+        response.headers[
+            "x-frame-options"
+        ]
+        == "DENY"
+    )
+
+
+def test_cors_accepts_openwebui_origin(
+    client: TestClient,
+) -> None:
+    response = client.options(
+        "/rag",
+        headers={
+            "Origin": (
+                "https://ui.tukekemt.xyz"
+            ),
+            "Access-Control-Request-Method": (
+                "POST"
+            ),
+            "Access-Control-Request-Headers": (
+                "authorization,"
+                "content-type,"
+                "x-api-key"
+            ),
+        },
+    )
+
+    assert (
+        response.status_code
+        == 200
+    )
+
+    assert response.headers[
+        "access-control-allow-origin"
+    ] == (
+        "https://ui.tukekemt.xyz"
+    )
+
+
+def test_cors_rejects_unknown_origin(
+    client: TestClient,
+) -> None:
+    response = client.options(
+        "/rag",
+        headers={
+            "Origin": (
+                "https://evil.example"
+            ),
+            "Access-Control-Request-Method": (
+                "POST"
+            ),
+            "Access-Control-Request-Headers": (
+                "authorization"
+            ),
+        },
+    )
+
+    assert (
+        response.headers.get(
+            "access-control-allow-origin"
+        )
+        != "https://evil.example"
+    )
+
+
+def test_webhook_rejects_oversized_payload(
+    client: TestClient,
+) -> None:
+    body = (
+        b"x"
+        * (
+            routes.MAX_WEBHOOK_BODY_BYTES
+            + 1
+        )
+    )
+
+    response = client.post(
+        "/webhook/gitea",
+        content=body,
+        headers={
+            "Content-Type": (
+                "application/json"
+            ),
+            "X-Gitea-Event": (
+                "push"
+            ),
+        },
+    )
+
+    assert (
+        response.status_code
+        == 413
+    )
+
+
+def test_webhook_requires_repository_full_name(
+    client: TestClient,
+) -> None:
+    body = json.dumps(
+        {
+            "repository": {
+                "name": (
+                    "zpwiki"
+                ),
+            }
+        }
+    ).encode(
+        "utf-8"
+    )
+
+    response = client.post(
+        "/webhook/gitea",
+        content=body,
+        headers={
+            "Content-Type": (
+                "application/json"
+            ),
+            "X-Gitea-Event": (
+                "push"
+            ),
+            "X-Gitea-Signature": (
+                sign(
+                    body
+                )
+            ),
+        },
+    )
+
+    assert (
+        response.status_code
+        == 400
+    )