mirror of
https://github.com/Quad4-Software/MeshChatX.git
synced 2026-08-18 09:49:09 -04:00
297 lines
11 KiB
Python
297 lines
11 KiB
Python
# SPDX-License-Identifier: 0BSD
|
|
|
|
import asyncio
|
|
import secrets
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
import bcrypt
|
|
import pytest
|
|
from aiohttp import web
|
|
from aiohttp.test_utils import TestClient, TestServer
|
|
from aiohttp_session import setup as setup_session
|
|
from hypothesis import HealthCheck, given, settings
|
|
from hypothesis import strategies as st
|
|
|
|
from meshchatx.src.backend.config_manager import ConfigManager
|
|
from meshchatx.src.backend.http.middleware import create_auth_middleware
|
|
from tests.backend.conftest import extend_meshchat_middlewares, fetch_api_csrf_headers
|
|
|
|
|
|
def _make_aio_app(mock_app, use_https: bool):
|
|
mock_app.session_secret_key = secrets.token_urlsafe(32)
|
|
mock_app.listen_host = "127.0.0.1"
|
|
mock_app.listen_port = 8000
|
|
mock_app.use_https = use_https
|
|
mock_app.landlock_active = False
|
|
routes = web.RouteTableDef()
|
|
middlewares = mock_app._define_routes(routes)
|
|
aio_app = web.Application()
|
|
setup_session(aio_app, mock_app._encrypted_cookie_storage(use_https))
|
|
extend_meshchat_middlewares(aio_app, middlewares)
|
|
aio_app.add_routes(routes)
|
|
return aio_app
|
|
|
|
|
|
def test_config_password_hash_roundtrip(mock_app):
|
|
mock_app.config.auth_password_hash.set("roundtrip-test-hash")
|
|
assert mock_app.config.auth_password_hash.get() == "roundtrip-test-hash"
|
|
|
|
|
|
def test_config_bcrypt_hash_roundtrip(mock_app):
|
|
mock_app.config.auth_enabled.set(True)
|
|
h = bcrypt.hashpw(b"pw", bcrypt.gensalt()).decode("utf-8")
|
|
mock_app.config.auth_password_hash.set(h)
|
|
assert mock_app.config.auth_password_hash.get() == h
|
|
|
|
|
|
def test_encrypted_cookie_storage_https_flags(mock_app):
|
|
mock_app.session_secret_key = secrets.token_urlsafe(32)
|
|
storage = mock_app._encrypted_cookie_storage(True)
|
|
assert storage.cookie_params["secure"] is True
|
|
assert storage.cookie_params["httponly"] is True
|
|
assert storage.cookie_params["samesite"] == "Lax"
|
|
|
|
|
|
def test_encrypted_cookie_storage_http_flags(mock_app):
|
|
mock_app.session_secret_key = secrets.token_urlsafe(32)
|
|
storage = mock_app._encrypted_cookie_storage(False)
|
|
assert storage.cookie_params["secure"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.usefixtures("require_loopback_tcp")
|
|
async def test_login_sets_cookie_and_allows_protected_api(mock_app):
|
|
mock_app.config.auth_enabled.set(True)
|
|
pw = b"integration-test-password-ok"
|
|
stored_hash = bcrypt.hashpw(pw, bcrypt.gensalt()).decode("utf-8")
|
|
mock_app.config.auth_password_hash.set(stored_hash)
|
|
assert isinstance(mock_app.current_context.config, ConfigManager)
|
|
assert mock_app.config.auth_password_hash.get() == stored_hash
|
|
aio_app = _make_aio_app(mock_app, use_https=False)
|
|
|
|
async with TestClient(TestServer(aio_app)) as client:
|
|
headers = await fetch_api_csrf_headers(client)
|
|
login = await client.post(
|
|
"/api/v1/auth/login",
|
|
json={"password": pw.decode("utf-8")},
|
|
headers=headers,
|
|
)
|
|
assert login.status == 200
|
|
set_cookie = login.headers.get("Set-Cookie", "")
|
|
assert "HttpOnly" in set_cookie
|
|
assert "SameSite=Lax" in set_cookie
|
|
assert "Secure" not in set_cookie
|
|
|
|
backups = await client.get("/api/v1/database/backups")
|
|
assert backups.status == 200
|
|
body = await backups.json()
|
|
assert "backups" in body
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.usefixtures("require_loopback_tcp")
|
|
async def test_ws_returns_401_without_session_when_auth_enabled(mock_app):
|
|
mock_app.config.auth_enabled.set(True)
|
|
mock_app.config.auth_password_hash.set(
|
|
bcrypt.hashpw(b"x", bcrypt.gensalt()).decode("utf-8"),
|
|
)
|
|
aio_app = _make_aio_app(mock_app, use_https=False)
|
|
|
|
async with TestClient(TestServer(aio_app)) as client:
|
|
r = await client.get("/ws")
|
|
assert r.status == 401
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.usefixtures("require_loopback_tcp")
|
|
async def test_logout_clears_session_for_protected_api(mock_app):
|
|
mock_app.config.auth_enabled.set(True)
|
|
pw = b"logout-test-password-ok"
|
|
mock_app.config.auth_password_hash.set(
|
|
bcrypt.hashpw(pw, bcrypt.gensalt()).decode("utf-8"),
|
|
)
|
|
aio_app = _make_aio_app(mock_app, use_https=False)
|
|
|
|
async with TestClient(TestServer(aio_app)) as client:
|
|
headers = await fetch_api_csrf_headers(client)
|
|
await client.post(
|
|
"/api/v1/auth/login",
|
|
json={"password": pw.decode("utf-8")},
|
|
headers=headers,
|
|
)
|
|
assert (await client.get("/api/v1/database/backups")).status == 200
|
|
|
|
headers = await fetch_api_csrf_headers(client)
|
|
out = await client.post("/api/v1/auth/logout", headers=headers)
|
|
assert out.status == 200
|
|
|
|
assert (await client.get("/api/v1/database/backups")).status == 401
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.usefixtures("require_loopback_tcp")
|
|
async def test_auth_login_invalid_json_returns_400(mock_app):
|
|
mock_app.config.auth_enabled.set(True)
|
|
mock_app.config.auth_password_hash.set(
|
|
bcrypt.hashpw(b"x", bcrypt.gensalt()).decode("utf-8"),
|
|
)
|
|
aio_app = _make_aio_app(mock_app, use_https=False)
|
|
|
|
async with TestClient(TestServer(aio_app)) as client:
|
|
headers = await fetch_api_csrf_headers(client)
|
|
r = await client.post(
|
|
"/api/v1/auth/login",
|
|
data="{not-json",
|
|
headers={**headers, "Content-Type": "application/json"},
|
|
)
|
|
assert r.status == 400
|
|
body = await r.json()
|
|
assert "error" in body
|
|
|
|
|
|
@settings(
|
|
max_examples=40,
|
|
suppress_health_check=[HealthCheck.function_scoped_fixture],
|
|
deadline=None,
|
|
)
|
|
@given(body=st.binary(min_size=0, max_size=12000))
|
|
@pytest.mark.usefixtures("require_loopback_tcp")
|
|
def test_auth_login_fuzz_never_500(mock_app, body):
|
|
mock_app.config.auth_enabled.set(True)
|
|
mock_app.config.auth_password_hash.set(
|
|
bcrypt.hashpw(b"fixed", bcrypt.gensalt()).decode("utf-8"),
|
|
)
|
|
aio_app = _make_aio_app(mock_app, use_https=False)
|
|
|
|
async def run():
|
|
async with TestClient(TestServer(aio_app)) as client:
|
|
headers = await fetch_api_csrf_headers(client)
|
|
r = await client.post(
|
|
"/api/v1/auth/login",
|
|
data=body,
|
|
headers={**headers, "Content-Type": "application/json"},
|
|
)
|
|
assert r.status != 500
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_reset_password_clears_hash_when_set(mock_app):
|
|
mock_app.config.auth_enabled.set(True)
|
|
h = bcrypt.hashpw(b"old-password", bcrypt.gensalt()).decode("utf-8")
|
|
mock_app.config.auth_password_hash.set(h)
|
|
assert mock_app.reset_password() is True
|
|
assert mock_app.config.auth_password_hash.get() is None
|
|
|
|
|
|
def test_reset_password_no_op_when_no_hash(mock_app):
|
|
mock_app.config.auth_password_hash.set(None)
|
|
assert mock_app.reset_password() is False
|
|
assert mock_app.config.auth_password_hash.get() is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.usefixtures("require_loopback_tcp")
|
|
async def test_reset_password_exposes_setup_screen(mock_app):
|
|
mock_app.config.auth_enabled.set(True)
|
|
h = bcrypt.hashpw(b"old-password", bcrypt.gensalt()).decode("utf-8")
|
|
mock_app.config.auth_password_hash.set(h)
|
|
assert mock_app.reset_password() is True
|
|
|
|
aio_app = _make_aio_app(mock_app, use_https=False)
|
|
async with TestClient(TestServer(aio_app)) as client:
|
|
status = await client.get("/api/v1/auth/status")
|
|
assert status.status == 200
|
|
body = await status.json()
|
|
assert body["password_set"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.usefixtures("require_loopback_tcp")
|
|
async def test_auth_rejects_api_paths_that_end_with_static_extensions(mock_app):
|
|
"""Unauthenticated /api/v1 paths must 401 even when the URL looks like a static file.
|
|
|
|
Plugin assets are /api/v1/plugins/{id}/asset/{path} and commonly end in
|
|
.js/.json/.wasm. Those suffixes are public only for non-API static files.
|
|
"""
|
|
mock_app.config.auth_enabled.set(True)
|
|
mock_app.config.auth_password_hash.set(
|
|
bcrypt.hashpw(b"x", bcrypt.gensalt()).decode("utf-8"),
|
|
)
|
|
aio_app = _make_aio_app(mock_app, use_https=False)
|
|
|
|
async with TestClient(TestServer(aio_app)) as client:
|
|
config = await client.get("/api/v1/config")
|
|
assert config.status == 401
|
|
|
|
disguised = await client.get("/api/v1/config.json")
|
|
assert disguised.status == 401, (
|
|
f"expected 401 for disguised API path, got {disguised.status}"
|
|
)
|
|
|
|
plugin_js = await client.get(
|
|
"/api/v1/plugins/com.meshchatx.mcx-bugs/asset/frontend/main.js",
|
|
)
|
|
assert plugin_js.status == 401, (
|
|
f"expected 401 for plugin JS asset, got {plugin_js.status}"
|
|
)
|
|
|
|
plugin_wasm = await client.get(
|
|
"/api/v1/plugins/com.meshchatx.mcx-bugs/asset/backend/main.wasm",
|
|
)
|
|
assert plugin_wasm.status == 401, (
|
|
f"expected 401 for plugin wasm asset, got {plugin_wasm.status}"
|
|
)
|
|
|
|
status_prefix = await client.get("/api/v1/status.json")
|
|
assert status_prefix.status == 401, (
|
|
f"expected 401 for status prefix disguise, got {status_prefix.status}"
|
|
)
|
|
|
|
# Do not GET /manifest.json here. That route uses FileResponse, which
|
|
# calls loop.run_in_executor. mock_app patches threading.Thread so the
|
|
# executor never runs and the client hangs until TimeoutError.
|
|
# A missing static asset is enough: middleware must not 401 it.
|
|
static_js = await client.get("/assets/does-not-exist.js")
|
|
assert static_js.status != 401, (
|
|
f"expected static non-API path to skip auth, got {static_js.status}"
|
|
)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_auth_middleware_does_not_treat_api_static_suffixes_as_public():
|
|
"""Oracle: password auth still applies when an API path ends in .js/.json/.wasm."""
|
|
app = MagicMock()
|
|
app.auth_enabled = True
|
|
app.current_context = MagicMock(running=True)
|
|
app.identity.hash.hex.return_value = "aa" * 16
|
|
app._startup_stage = "ok"
|
|
|
|
handler = AsyncMock(return_value=web.Response(status=200, text="ok"))
|
|
mw = create_auth_middleware(app)
|
|
empty_session = AsyncMock(return_value={})
|
|
|
|
async def call(path):
|
|
request = MagicMock()
|
|
request.path = path
|
|
with patch(
|
|
"meshchatx.src.backend.http.middleware.get_session",
|
|
empty_session,
|
|
):
|
|
return await mw(request, handler)
|
|
|
|
for path in (
|
|
"/api/v1/config.json",
|
|
"/api/v1/status.json",
|
|
"/api/v1/plugins/com.meshchatx.mcx-bugs/asset/frontend/main.js",
|
|
"/api/v1/plugins/com.meshchatx.mcx-bugs/asset/backend/main.wasm",
|
|
):
|
|
handler.reset_mock()
|
|
resp = await call(path)
|
|
assert resp.status == 401, f"{path} expected 401, got {resp.status}"
|
|
handler.assert_not_awaited()
|
|
|
|
handler.reset_mock()
|
|
public = await call("/manifest.json")
|
|
assert public.status == 200
|
|
handler.assert_awaited_once()
|