443 lines
8.7 KiB
Python
443 lines
8.7 KiB
Python
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
|