61ed636df0
Co-authored-by: alan <alan@lnbits.com> Co-authored-by: Tiago Vasconcelos <talvasconcelos@gmail.com>
280 lines
9.1 KiB
Python
280 lines
9.1 KiB
Python
from __future__ import annotations
|
|
|
|
from collections.abc import AsyncIterator
|
|
from pathlib import Path
|
|
from typing import cast
|
|
|
|
import pytest
|
|
from fastapi import HTTPException, Request
|
|
|
|
from lnbits.core.wasm_ext.routes.api import (
|
|
WasmRequestBodyTooLargeError,
|
|
_read_api_payload,
|
|
_read_json_object_with_size,
|
|
_wasm_extension_api_export,
|
|
)
|
|
from lnbits.core.wasm_ext.routes.assets import (
|
|
_reject_html_like_wasm_static_asset,
|
|
_wasm_extension_core_asset_response,
|
|
)
|
|
from lnbits.core.wasm_ext.routes.security import (
|
|
consume_wasm_extension_frame_token,
|
|
wasm_extension_frame_csp,
|
|
wasm_extension_frame_url,
|
|
)
|
|
from lnbits.core.wasm_ext.routes.ui import (
|
|
_match_wasm_extension_ui_route,
|
|
_wasm_extension_bridge_api_routes,
|
|
_wasm_extension_entrypoint,
|
|
)
|
|
from lnbits.core.wasm_ext.wasm.config import parse_wasm_extension_config
|
|
from lnbits.core.wasm_ext.wasm.loader import WasmExtension
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_wasm_json_reader_rejects_large_content_length_without_reading():
|
|
request = _FakeRequest([b"{}"], content_length="11")
|
|
|
|
with pytest.raises(WasmRequestBodyTooLargeError, match="11 bytes"):
|
|
await _read_json_object_with_size(cast(Request, request), max_body_bytes=10)
|
|
|
|
assert request.stream_started is False
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_wasm_json_reader_rejects_large_stream_without_content_length():
|
|
request = _FakeRequest([b'{"value":"', b"x" * 20, b'"}'])
|
|
|
|
with pytest.raises(WasmRequestBodyTooLargeError):
|
|
await _read_json_object_with_size(cast(Request, request), max_body_bytes=16)
|
|
|
|
assert request.stream_started is True
|
|
|
|
|
|
@pytest.mark.anyio
|
|
async def test_wasm_api_payload_records_actual_body_bytes():
|
|
body = b'{"amount":21}'
|
|
request = _FakeRequest(
|
|
[body],
|
|
path_params={"invoice_id": "abc"},
|
|
query_params={"include_paid": "true"},
|
|
)
|
|
|
|
payload = await _read_api_payload(
|
|
cast(Request, request),
|
|
{"invoice_id": "invoiceId"},
|
|
max_body_bytes=100,
|
|
)
|
|
|
|
assert payload.data == {
|
|
"invoiceId": "abc",
|
|
"includePaid": "true",
|
|
"amount": 21,
|
|
}
|
|
assert payload.request_bytes == len(body)
|
|
|
|
|
|
def test_wasm_api_export_visibility_is_enforced(tmp_path: Path):
|
|
extension = _wasm_extension(tmp_path)
|
|
|
|
assert _wasm_extension_api_export(extension, "render") == "render"
|
|
assert _wasm_extension_api_export(extension, "private_render") == "private_render"
|
|
with pytest.raises(PermissionError, match="not callable over HTTP"):
|
|
_wasm_extension_api_export(extension, "on_invoice_paid")
|
|
with pytest.raises(KeyError, match="has no export"):
|
|
_wasm_extension_api_export(extension, "missing")
|
|
|
|
|
|
def test_wasm_ui_entrypoint_rejects_escape_static_and_non_html(tmp_path: Path):
|
|
extension = _wasm_extension(tmp_path)
|
|
(tmp_path / "index.html").write_text("<html></html>", encoding="utf-8")
|
|
(tmp_path / "index.txt").write_text("text", encoding="utf-8")
|
|
(tmp_path / "static").mkdir()
|
|
(tmp_path / "static" / "index.html").write_text("<html></html>", encoding="utf-8")
|
|
|
|
assert (
|
|
_wasm_extension_entrypoint(extension, "index.html")
|
|
== (tmp_path / "index.html").resolve()
|
|
)
|
|
with pytest.raises(ValueError, match="escapes extension root"):
|
|
_wasm_extension_entrypoint(extension, "../outside.html")
|
|
with pytest.raises(ValueError, match="must not be inside the static"):
|
|
_wasm_extension_entrypoint(extension, "static/index.html")
|
|
with pytest.raises(ValueError, match="must be an HTML file"):
|
|
_wasm_extension_entrypoint(extension, "index.txt")
|
|
|
|
|
|
def test_wasm_frame_token_is_one_time_and_user_bound(tmp_path: Path):
|
|
extension = _wasm_extension(tmp_path)
|
|
frame_path = "/ext-frame/demoext/0"
|
|
frame_url = wasm_extension_frame_url(extension, frame_path, "user-1")
|
|
token = frame_url.split("frame_token=", 1)[1]
|
|
|
|
with pytest.raises(HTTPException) as wrong_user:
|
|
consume_wasm_extension_frame_token(
|
|
_request_with_query(token),
|
|
extension,
|
|
frame_path,
|
|
"user-2",
|
|
)
|
|
assert wrong_user.value.status_code == 404
|
|
|
|
consume_wasm_extension_frame_token(
|
|
_request_with_query(token),
|
|
extension,
|
|
frame_path,
|
|
"user-1",
|
|
)
|
|
with pytest.raises(HTTPException) as reused:
|
|
consume_wasm_extension_frame_token(
|
|
_request_with_query(token),
|
|
extension,
|
|
frame_path,
|
|
"user-1",
|
|
)
|
|
assert reused.value.status_code == 404
|
|
|
|
|
|
def test_wasm_frame_csp_is_locked_to_extension_assets(tmp_path: Path):
|
|
csp = wasm_extension_frame_csp(
|
|
_request_with_query("token"),
|
|
_wasm_extension(tmp_path),
|
|
)
|
|
|
|
assert "sandbox allow-scripts" in csp
|
|
assert "default-src 'none'" in csp
|
|
assert "connect-src 'none'" in csp
|
|
assert "frame-ancestors 'self'" in csp
|
|
assert "http://testserver/ext-assets/demoext/" in csp
|
|
|
|
|
|
def test_wasm_ui_route_matching_and_bridge_public_api_filtering(tmp_path: Path):
|
|
extension = _wasm_extension(tmp_path)
|
|
|
|
matched = _match_wasm_extension_ui_route(extension, "/ext/demo/abc")
|
|
public_routes = _wasm_extension_bridge_api_routes(extension, public=True)
|
|
private_routes = _wasm_extension_bridge_api_routes(extension, public=False)
|
|
|
|
assert matched["auth"] == "user"
|
|
assert matched["route_params"] == {"item_id": "abc"}
|
|
assert public_routes == [
|
|
{
|
|
"method": "GET",
|
|
"path": "/api/v1/ext/demoext/public/{item_id}",
|
|
"pattern": "^/api/v1/ext/demoext/public/[^/]+$",
|
|
}
|
|
]
|
|
assert {route["path"] for route in private_routes} == {
|
|
"/api/v1/ext/demoext/public/{item_id}",
|
|
"/api/v1/ext/demoext/private/{item_id}",
|
|
}
|
|
|
|
|
|
def test_wasm_static_core_assets_and_html_like_text_assets_are_guarded(tmp_path: Path):
|
|
response = _wasm_extension_core_asset_response("_lnbits/material-icons.css")
|
|
assert response.headers["X-Content-Type-Options"] == "nosniff"
|
|
assert response.headers["Cache-Control"] == "no-store"
|
|
|
|
for path in ["_lnbits/../bundle.min.css", "_lnbits/missing.css"]:
|
|
with pytest.raises(HTTPException) as exc_info:
|
|
_wasm_extension_core_asset_response(path)
|
|
assert exc_info.value.status_code == 404
|
|
|
|
script_path = tmp_path / "app.js"
|
|
script_path.write_text("<script>alert(1)</script>", encoding="utf-8")
|
|
with pytest.raises(HTTPException) as html_like:
|
|
_reject_html_like_wasm_static_asset(script_path)
|
|
assert html_like.value.status_code == 404
|
|
|
|
|
|
class _FakeRequest:
|
|
method = "POST"
|
|
|
|
def __init__(
|
|
self,
|
|
chunks: list[bytes],
|
|
*,
|
|
content_length: str | None = None,
|
|
path_params: dict[str, str] | None = None,
|
|
query_params: dict[str, str] | None = None,
|
|
) -> None:
|
|
self._chunks = chunks
|
|
self.headers: dict[str, str] = {}
|
|
if content_length is not None:
|
|
self.headers["content-length"] = content_length
|
|
self.path_params = path_params or {}
|
|
self.query_params = query_params or {}
|
|
self.stream_started = False
|
|
|
|
async def stream(self) -> AsyncIterator[bytes]:
|
|
self.stream_started = True
|
|
for chunk in self._chunks:
|
|
yield chunk
|
|
|
|
|
|
def _wasm_extension(root_path: Path) -> WasmExtension:
|
|
config = parse_wasm_extension_config(
|
|
"demoext",
|
|
{
|
|
"id": "demoext",
|
|
"name": "Demo",
|
|
"short_description": "Demo extension",
|
|
"version": "1.0.0",
|
|
"extension_type": "wasm",
|
|
"wasm": {
|
|
"module": "extension.wasm",
|
|
"exports": [
|
|
{"name": "render", "visibility": "public"},
|
|
{"name": "private_render", "visibility": "authenticated"},
|
|
{"name": "on_invoice_paid", "visibility": "event"},
|
|
],
|
|
},
|
|
"ui_routes": [
|
|
{
|
|
"path": "/demo/{item_id}",
|
|
"entrypoint": "index.html",
|
|
"auth": "user",
|
|
}
|
|
],
|
|
"api_routes": [
|
|
{
|
|
"method": "GET",
|
|
"path": "/public/{item_id}",
|
|
"export": "render",
|
|
"auth": "public",
|
|
},
|
|
{
|
|
"method": "POST",
|
|
"path": "/private/{item_id}",
|
|
"export": "private_render",
|
|
"auth": "user",
|
|
},
|
|
],
|
|
},
|
|
)
|
|
return WasmExtension(
|
|
id="demoext",
|
|
name="Demo",
|
|
version="1.0.0",
|
|
root_path=root_path,
|
|
module_path=root_path / "extension.wasm",
|
|
wit_path=None,
|
|
world="",
|
|
exports=config.wasm.exports,
|
|
config=config,
|
|
)
|
|
|
|
|
|
def _request_with_query(token: str) -> Request:
|
|
return Request(
|
|
{
|
|
"type": "http",
|
|
"method": "GET",
|
|
"scheme": "http",
|
|
"server": ("testserver", 80),
|
|
"path": "/ext-frame/demoext/0",
|
|
"query_string": f"frame_token={token}".encode(),
|
|
"headers": [],
|
|
}
|
|
)
|