255 lines
10 KiB
Python
255 lines
10 KiB
Python
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import os
|
|
import sys
|
|
import unittest
|
|
import urllib.parse
|
|
from pathlib import Path
|
|
|
|
|
|
SOURCE = Path(__file__).parents[1] / "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):
|
|
@staticmethod
|
|
def dns_form(*records: tuple[str, str, str]) -> str:
|
|
fields = []
|
|
for record_type, prefix, value in records:
|
|
fields.append(
|
|
f'<select name="type"><option value="{record_type}" selected>'
|
|
f'{record_type}</option></select>'
|
|
f'<input name="prefix" value="{prefix}">'
|
|
f'<textarea name="value">{value}</textarea>'
|
|
)
|
|
fields.append('<input name="action_change_txt_records" value="Speichern">')
|
|
return "".join(fields)
|
|
|
|
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>
|
|
<input name="action_change_txt_records" value="Speichern">
|
|
"""
|
|
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 = self.dns_form(("CNAME", "media", "proxy.example.net."))
|
|
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)
|
|
|
|
def client_for_write(self, responses: list[FakeResponse]) -> tuple[object, FakeOpener]:
|
|
opener = FakeOpener(responses)
|
|
config = module.StratoConfig(
|
|
"customer", "secret-value", "example.de",
|
|
package_id="42", write_enabled=True,
|
|
)
|
|
client = module.StratoClient(config, opener=opener, sleep=lambda _seconds: None)
|
|
client.session_id = "test-session"
|
|
return client, opener
|
|
|
|
@staticmethod
|
|
def posted_form(opener: FakeOpener) -> dict[str, list[str]]:
|
|
posts = [request for request, _timeout in opener.requests if request.method == "POST"]
|
|
assert len(posts) == 1
|
|
return urllib.parse.parse_qs(posts[0].data.decode(), keep_blank_values=True)
|
|
|
|
def test_create_cname_preserves_all_existing_records_and_verifies(self) -> None:
|
|
before = self.dns_form(
|
|
("TXT", "_acme-challenge", "validation-value"),
|
|
("CNAME", "media", "proxy.example.net."),
|
|
)
|
|
after = self.dns_form(
|
|
("TXT", "_acme-challenge", "validation-value"),
|
|
("CNAME", "media", "proxy.example.net."),
|
|
("CNAME", "chat", "gateway.example.net"),
|
|
)
|
|
client, opener = self.client_for_write([
|
|
FakeResponse(module.STRATO_URL, before),
|
|
FakeResponse(module.STRATO_URL, "saved"),
|
|
FakeResponse(module.STRATO_URL, after),
|
|
])
|
|
|
|
created = client.create_cname("chat.example.de", "gateway.example.net.")
|
|
|
|
self.assertEqual(created.as_dict(), {
|
|
"type": "CNAME", "prefix": "chat", "value": "gateway.example.net"
|
|
})
|
|
posted = self.posted_form(opener)
|
|
self.assertEqual(posted["type"], ["TXT", "CNAME", "CNAME"])
|
|
self.assertEqual(posted["prefix"], ["_acme-challenge", "media", "chat"])
|
|
self.assertEqual(
|
|
posted["value"],
|
|
["validation-value", "proxy.example.net.", "gateway.example.net"],
|
|
)
|
|
self.assertEqual(posted["action_change_txt_records"], ["Speichern"])
|
|
|
|
def test_update_cname_can_change_target_and_prefix(self) -> None:
|
|
before = self.dns_form(
|
|
("TXT", "verification", "keep-me"),
|
|
("CNAME", "old", "old-target.example.net"),
|
|
)
|
|
after = self.dns_form(
|
|
("TXT", "verification", "keep-me"),
|
|
("CNAME", "new", "new-target.example.net"),
|
|
)
|
|
client, opener = self.client_for_write([
|
|
FakeResponse(module.STRATO_URL, before),
|
|
FakeResponse(module.STRATO_URL, "saved"),
|
|
FakeResponse(module.STRATO_URL, after),
|
|
])
|
|
|
|
old, new = client.update_cname("old", "new-target.example.net", new_prefix="new")
|
|
|
|
self.assertEqual(old.prefix, "old")
|
|
self.assertEqual(new.prefix, "new")
|
|
posted = self.posted_form(opener)
|
|
self.assertEqual(posted["prefix"], ["verification", "new"])
|
|
self.assertEqual(posted["value"], ["keep-me", "new-target.example.net"])
|
|
|
|
def test_delete_cname_preserves_unrelated_records_and_verifies(self) -> None:
|
|
before = self.dns_form(
|
|
("TXT", "verification", "keep-me"),
|
|
("CNAME", "remove", "old.example.net"),
|
|
("CNAME", "keep", "keep.example.net"),
|
|
)
|
|
after = self.dns_form(
|
|
("TXT", "verification", "keep-me"),
|
|
("CNAME", "keep", "keep.example.net"),
|
|
)
|
|
client, opener = self.client_for_write([
|
|
FakeResponse(module.STRATO_URL, before),
|
|
FakeResponse(module.STRATO_URL, "saved"),
|
|
FakeResponse(module.STRATO_URL, after),
|
|
])
|
|
|
|
deleted = client.delete_cname("remove")
|
|
|
|
self.assertEqual(deleted.prefix, "remove")
|
|
posted = self.posted_form(opener)
|
|
self.assertEqual(posted["prefix"], ["verification", "keep"])
|
|
self.assertEqual(posted["value"], ["keep-me", "keep.example.net"])
|
|
|
|
def test_write_is_blocked_without_explicit_enable_flag(self) -> None:
|
|
opener = FakeOpener([
|
|
FakeResponse(module.STRATO_URL, self.dns_form(
|
|
("CNAME", "media", "proxy.example.net")
|
|
)),
|
|
])
|
|
config = module.StratoConfig(
|
|
"customer", "secret-value", "example.de", package_id="42"
|
|
)
|
|
client = module.StratoClient(config, opener=opener)
|
|
client.session_id = "test-session"
|
|
|
|
with self.assertRaises(module.StratoWriteDisabledError):
|
|
client.delete_cname("media")
|
|
self.assertEqual([request.method for request, _ in opener.requests], ["GET"])
|
|
|
|
def test_invalid_names_and_collisions_fail_before_post(self) -> None:
|
|
with self.assertRaises(module.StratoError):
|
|
module.normalize_cname_prefix("@", "example.de")
|
|
with self.assertRaises(module.StratoError):
|
|
module.normalize_cname_prefix("foo.*", "example.de")
|
|
with self.assertRaises(module.StratoError):
|
|
module.normalize_cname_target("https://target.example.net/path")
|
|
with self.assertRaises(module.StratoError):
|
|
module.normalize_cname_target("target..example.net")
|
|
|
|
client, opener = self.client_for_write([
|
|
FakeResponse(module.STRATO_URL, self.dns_form(
|
|
("TXT", "chat", "already-used")
|
|
)),
|
|
])
|
|
with self.assertRaisesRegex(module.StratoError, "already exists"):
|
|
client.create_cname("chat", "target.example.net")
|
|
self.assertEqual([request.method for request, _ in opener.requests], ["GET"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|