92 lines
2.7 KiB
Python
92 lines
2.7 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from k1link.device_plugins.xgrids_k1.facade import XGRIDS_K1_PLUGIN_ID
|
|
from k1link.web.app import INVALID_REQUEST_DETAIL, app
|
|
|
|
|
|
async def _post_json(path: str, payload: dict[str, Any]) -> tuple[int, str]:
|
|
body = json.dumps(payload).encode()
|
|
request_sent = False
|
|
response_messages: list[dict[str, Any]] = []
|
|
|
|
async def receive() -> dict[str, Any]:
|
|
nonlocal request_sent
|
|
if request_sent:
|
|
return {"type": "http.disconnect"}
|
|
request_sent = True
|
|
return {"type": "http.request", "body": body, "more_body": False}
|
|
|
|
async def send(message: dict[str, Any]) -> None:
|
|
response_messages.append(message)
|
|
|
|
scope: dict[str, Any] = {
|
|
"type": "http",
|
|
"asgi": {"version": "3.0", "spec_version": "2.3"},
|
|
"http_version": "1.1",
|
|
"method": "POST",
|
|
"scheme": "http",
|
|
"path": path,
|
|
"raw_path": path.encode(),
|
|
"query_string": b"",
|
|
"root_path": "",
|
|
"headers": [
|
|
(b"content-type", b"application/json"),
|
|
(b"content-length", str(len(body)).encode()),
|
|
],
|
|
"client": ("127.0.0.1", 41000),
|
|
"server": ("127.0.0.1", 8765),
|
|
}
|
|
await app(scope, receive, send)
|
|
|
|
start = next(
|
|
message for message in response_messages if message["type"] == "http.response.start"
|
|
)
|
|
response_body = b"".join(
|
|
message.get("body", b"")
|
|
for message in response_messages
|
|
if message["type"] == "http.response.body"
|
|
)
|
|
return int(start["status"]), response_body.decode()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("path", "wrap_input"),
|
|
[
|
|
(
|
|
f"/api/v1/device-plugins/{XGRIDS_K1_PLUGIN_ID}/actions/network.provision",
|
|
True,
|
|
),
|
|
("/api/connect", False),
|
|
],
|
|
)
|
|
def test_validation_errors_do_not_echo_sensitive_request_values(
|
|
path: str,
|
|
wrap_input: bool,
|
|
) -> None:
|
|
sensitive_value = "x" * 300
|
|
action_input = {
|
|
"device_id": "synthetic-device",
|
|
"ssid": "synthetic-network",
|
|
"password": sensitive_value,
|
|
"compatibility_attestation": {
|
|
"firmware_version": "3.0.2",
|
|
"topology": "direct-lan",
|
|
"verification": "live-device-info",
|
|
},
|
|
}
|
|
payload = {"input": action_input} if wrap_input else action_input
|
|
|
|
status_code, response_text = asyncio.run(_post_json(path, payload))
|
|
|
|
assert status_code == 422
|
|
assert json.loads(response_text) == {"detail": INVALID_REQUEST_DETAIL}
|
|
assert sensitive_value not in response_text
|
|
assert sensitive_value[:32] not in response_text
|
|
assert "input_value" not in response_text
|