117 lines
4.6 KiB
Python
117 lines
4.6 KiB
Python
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import os
|
|
import sys
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
|
|
SOURCE = Path(__file__).parents[1] / "services/strato-dns-mcp/strato_client.py"
|
|
spec = importlib.util.spec_from_file_location("strato_client", SOURCE)
|
|
module = importlib.util.module_from_spec(spec)
|
|
assert spec.loader
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
|
|
|
|
class FakeResponse:
|
|
def __init__(self, url: str, body: str) -> None:
|
|
self.url = url
|
|
self.body = body.encode()
|
|
|
|
def read(self, _limit: int) -> bytes:
|
|
return self.body
|
|
|
|
def geturl(self) -> str:
|
|
return self.url
|
|
|
|
|
|
class FakeOpener:
|
|
def __init__(self, responses: list[FakeResponse]) -> None:
|
|
self.responses = responses
|
|
self.requests: list[tuple[object, float]] = []
|
|
|
|
def open(self, request: object, timeout: float) -> FakeResponse:
|
|
self.requests.append((request, timeout))
|
|
return self.responses.pop(0)
|
|
|
|
|
|
class StratoParserTests(unittest.TestCase):
|
|
def test_dns_form_is_parsed_without_script_or_markup(self) -> None:
|
|
html = """
|
|
<select name="type"><option value="TXT">TXT</option><option value="CNAME" selected>CNAME</option></select>
|
|
<input name="prefix" value="media">
|
|
<textarea name="value">proxy.example.net.</textarea>
|
|
<select name="type"><option value="TXT" selected>TXT</option></select>
|
|
<input name="prefix" value="_acme-challenge">
|
|
<textarea name="value">public-validation-value</textarea>
|
|
"""
|
|
records = module.parse_records(html)
|
|
self.assertEqual(records[0].as_dict(), {
|
|
"type": "CNAME", "prefix": "media", "value": "proxy.example.net."
|
|
})
|
|
self.assertEqual(records[1].type, "TXT")
|
|
|
|
def test_changed_form_fails_closed(self) -> None:
|
|
with self.assertRaisesRegex(module.StratoParseError, "field counts"):
|
|
module.parse_records('<input name="prefix" value="orphan">')
|
|
|
|
def test_package_is_selected_by_domain_not_fallback(self) -> None:
|
|
html = """
|
|
<table id="package_list"><tr><td>other.example</td><td><a href="?cID=1">open</a></td></tr>
|
|
<tr><td>example.de Hosting</td><td><a href="?node=x&cID=42">open</a></td></tr></table>
|
|
"""
|
|
self.assertEqual(module.parse_package_id(html, "example.de"), "42")
|
|
with self.assertRaises(module.StratoParseError):
|
|
module.parse_package_id(html, "missing.de")
|
|
|
|
def test_totp_matches_rfc_vector_truncated_to_six_digits(self) -> None:
|
|
# RFC 6238 secret, SHA-1, at t=59 gives 94287082 (therefore 287082 for 6 digits).
|
|
secret = "GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ"
|
|
self.assertEqual(module._totp(secret, at_time=59), "287082")
|
|
|
|
def test_environment_errors_do_not_echo_values(self) -> None:
|
|
old = dict(os.environ)
|
|
try:
|
|
for key in ("STRATO_USERNAME", "STRATO_PASSWORD", "STRATO_DOMAIN"):
|
|
os.environ.pop(key, None)
|
|
with self.assertRaises(module.StratoError) as caught:
|
|
module.StratoConfig.from_env()
|
|
message = str(caught.exception)
|
|
self.assertIn("STRATO_PASSWORD", message)
|
|
self.assertNotIn("secret-value", message)
|
|
finally:
|
|
os.environ.clear()
|
|
os.environ.update(old)
|
|
|
|
def test_complete_read_path_has_no_dns_write_request(self) -> None:
|
|
package_html = """
|
|
<table id="package_list"><tr><td>example.de Hosting</td>
|
|
<td><a href="?node=x&cID=42">open</a></td></tr></table>
|
|
"""
|
|
records_html = """
|
|
<select name="type"><option value="CNAME" selected>CNAME</option></select>
|
|
<input name="prefix" value="media">
|
|
<textarea name="value">proxy.example.net.</textarea>
|
|
"""
|
|
opener = FakeOpener([
|
|
FakeResponse(module.STRATO_URL, "login"),
|
|
FakeResponse(module.STRATO_URL + "?sessionID=test-session", "welcome"),
|
|
FakeResponse(module.STRATO_URL, package_html),
|
|
FakeResponse(module.STRATO_URL, records_html),
|
|
])
|
|
config = module.StratoConfig("customer", "secret-value", "example.de")
|
|
records = module.StratoClient(
|
|
config, opener=opener, sleep=lambda _seconds: None
|
|
).list_cnames()
|
|
self.assertEqual([record.prefix for record in records], ["media"])
|
|
methods = [request.method for request, _timeout in opener.requests]
|
|
self.assertEqual(methods, ["GET", "POST", "GET", "GET"])
|
|
self.assertNotIn("secret-value", opener.requests[1][0].full_url)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|