diff --git a/README.md b/README.md index 40ffe73..e44dcf6 100644 --- a/README.md +++ b/README.md @@ -1,57 +1,59 @@ -# STRATO-DNS-MCP – Read-only-Prototyp +# STRATO-DNS-MCP -Dieser Prototyp prüft den undokumentierten HTTP-Leseweg des STRATO- -Kundenbereichs. Er kann sich anmelden und die CNAME-Einträge genau einer -konfigurierten DNS-Zone auflisten. - -**Diese Version kann keine DNS-Einträge erstellen, ändern oder löschen.** Im -Quellcode existiert absichtlich keine Speichermethode. +Dieser MCP verwaltet die CNAME-Einträge genau einer konfigurierten DNS-Zone im +STRATO-Kundenbereich. Er kann Einträge auflisten, anlegen, ändern und löschen. ## Technische Grundlage STRATO dokumentiert für normale Hosting-Domains nur die Verwaltung im Kunden-Login. Der öffentliche Certbot-Plugin [`FlixMa/certbot-dns-strato`](https://github.com/FlixMa/certbot-dns-strato) -zeigt jedoch einen funktionsfähigen HTTP-Ablauf über -`https://www.strato.de/apps/CustomerService`. Der hier verwendete Leseweg wurde -ohne Certbot-Abhängigkeit neu und deutlich defensiver implementiert. +zeigt den von STRATO selbst verwendeten HTTP-Ablauf über +`https://www.strato.de/apps/CustomerService`. -Referenzstand der Untersuchung: +Der MCP liest vor jeder Änderung das komplette TXT/CNAME-Formular, verändert +genau den gewünschten CNAME, sendet alle übrigen Einträge unverändert zurück +und liest danach erneut zur Kontrolle. Referenzstand der Untersuchung: `FlixMa/certbot-dns-strato@67df6dcfc3ef0035ec5aa5daf7c5b0bd8310d7fb`. -## Lokaler Test – noch nicht produktiv installieren +## Werkzeuge + +- `strato_connection_status` +- `strato_list_cnames` +- `strato_create_cname` +- `strato_update_cname` +- `strato_delete_cname` + +## Konfiguration und Start 1. `strato.env.example` außerhalb von Git nach `strato.env` kopieren. -2. Dort Benutzername, Passwort und DNS-Zone eintragen. -3. Falls STRATO TOTP verlangt, zusätzlich TOTP-Secret und den bei STRATO - angezeigten Gerätenamen eintragen. -4. Image bauen und zunächst ausschließlich `strato_connection_status` sowie - `strato_list_cnames` testen. +2. Benutzername, Passwort und DNS-Zone eintragen. +3. Falls STRATO TOTP verlangt, TOTP-Secret und Gerätenamen eintragen. +4. Für Änderungen zusätzlich `STRATO_WRITE_ENABLED=true` setzen. Ohne diesen + Schalter bleiben die drei Schreibwerkzeuge gesperrt. -Das Docker-Image wird direkt aus diesem Repository gebaut: +Das Docker-Image wird direkt aus diesem Repository gebaut. Der historische +Image-Name bleibt erhalten, damit eine bestehende Unraid-Vorlage ohne Umbau +weiter funktioniert: ```sh docker build -t strato-dns-mcp:readonly . ``` -Die Offline-Tests benötigen keine Zugangsdaten und kontaktieren STRATO nicht: +Die Offline-Tests benötigen keine Zugangsdaten, kontaktieren STRATO nicht und +prüfen auch, dass bestehende TXT- und CNAME-Einträge erhalten bleiben: ```sh python3 -m unittest -v tests/test_strato_readonly.py ``` Zugangsdaten, TOTP-Werte, Cookies und STRATO-Session-ID werden weder geloggt -noch als Toolausgabe zurückgegeben. Fehlermeldungen enthalten nur eine kurze -Fehlerklasse. +noch als Toolausgabe zurückgegeben. -## Noch bewusst nicht enthalten +## Produktiver Funktionstest -- kein Compose-Deployment -- keine Unraid-XML -- keine Hermes-Registrierung -- keine schreibenden Werkzeuge -- keine produktive Installation - -Diese Teile kommen erst, wenn der Read-only-Test gegen das aktuelle STRATO- -Konto funktioniert. Falls sich der Login oder das HTML geändert hat, wird nur -der Parser angepasst; es findet kein Schreibversuch statt. +Echte DNS-Schreibtests werden nicht automatisch ausgeführt. Nach dem Neubau +des Containers sollte ein eigens dafür vorgesehener Testname einmal angelegt, +geändert und wieder gelöscht werden. Danach stehen die neuen Werkzeuge in +Hermes nach einer frischen MCP-Verbindung beziehungsweise einer neuen Sitzung +zur Verfügung. diff --git a/strato.env.example b/strato.env.example index 9e3ef99..590824a 100644 --- a/strato.env.example +++ b/strato.env.example @@ -10,4 +10,7 @@ STRATO_PACKAGE_ID= STRATO_TOTP_SECRET= STRATO_TOTP_DEVICE= +# Schreibwerkzeuge bleiben ohne diese bewusste Freigabe gesperrt. +STRATO_WRITE_ENABLED=false + STRATO_TIMEOUT_SECONDS=20 diff --git a/strato_client.py b/strato_client.py index 87ee00f..9644912 100644 --- a/strato_client.py +++ b/strato_client.py @@ -1,12 +1,11 @@ #!/usr/bin/env python3 -"""Small read-only HTTP client for STRATO's customer portal. +"""Small HTTP client for STRATO's customer portal. STRATO does not publish a DNS-zone API for ordinary hosted domains. This -client deliberately implements only the minimum read path proven by the -public certbot-dns-strato project: authenticate, resolve the package that owns -one configured DNS zone, and read its combined TXT/CNAME form. - -There is intentionally no method that submits DNS changes. +client implements the minimum path proven by the public certbot-dns-strato +project: authenticate, resolve the package that owns one configured DNS zone, +read its combined TXT/CNAME form, and replace that form after a narrowly scoped +CNAME change. Every write is followed by a fresh read for verification. """ from __future__ import annotations @@ -29,7 +28,7 @@ from typing import Callable, Iterable STRATO_URL = "https://www.strato.de/apps/CustomerService" MAX_RESPONSE_BYTES = 5 * 1024 * 1024 -USER_AGENT = "Mozilla/5.0 (compatible; mike-ai-strato-dns-readonly/0.1)" +USER_AGENT = "Mozilla/5.0 (compatible; mike-ai-strato-dns-mcp/0.2)" class StratoError(RuntimeError): @@ -44,6 +43,10 @@ class StratoParseError(StratoError): pass +class StratoWriteDisabledError(StratoError): + pass + + @dataclass(frozen=True) class StratoConfig: username: str @@ -53,6 +56,7 @@ class StratoConfig: totp_secret: str | None = None totp_device: str | None = None timeout_seconds: float = 20.0 + write_enabled: bool = False @classmethod def from_env(cls) -> "StratoConfig": @@ -77,6 +81,8 @@ class StratoConfig: totp_secret=os.environ.get("STRATO_TOTP_SECRET", "").strip() or None, totp_device=os.environ.get("STRATO_TOTP_DEVICE", "").strip() or None, timeout_seconds=float(os.environ.get("STRATO_TIMEOUT_SECONDS", "20")), + write_enabled=os.environ.get("STRATO_WRITE_ENABLED", "false").strip().lower() + in {"1", "true", "yes", "on"}, ) @@ -90,6 +96,12 @@ class DnsRecord: return {"type": self.type, "prefix": self.prefix, "value": self.value} +@dataclass(frozen=True) +class DnsForm: + records: tuple[DnsRecord, ...] + submit_value: str + + class _FormParser(HTMLParser): """Extract parallel type/prefix/value fields from STRATO's DNS form.""" @@ -103,11 +115,14 @@ class _FormParser(HTMLParser): self._option_value = "" self._in_value_textarea = False self._textarea_parts: list[str] = [] + self.submit_value: str | None = None def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None: data = dict(attrs) if tag == "input" and data.get("name") == "prefix": self.prefixes.append(data.get("value") or "") + elif tag == "input" and data.get("name") == "action_change_txt_records": + self.submit_value = data.get("value") or "" elif tag == "select" and data.get("name") == "type": self._in_type_select = True elif tag == "option" and self._in_type_select: @@ -176,7 +191,7 @@ def _totp(secret: str, at_time: int | None = None) -> str: return f"{number % 1_000_000:06d}" -def parse_records(html: str) -> list[DnsRecord]: +def parse_dns_form(html: str) -> DnsForm: parser = _FormParser() parser.feed(html) counts = (len(parser.types), len(parser.prefixes), len(parser.values)) @@ -184,9 +199,51 @@ def parse_records(html: str) -> list[DnsRecord]: raise StratoParseError( "STRATO DNS form changed: type/prefix/value field counts do not match" ) - return [DnsRecord(t.upper(), p.strip(), v.strip()) for t, p, v in zip( + if not parser.submit_value: + raise StratoParseError("STRATO DNS form changed: submit action is missing") + records = tuple(DnsRecord(t.upper(), p.strip(), v.strip()) for t, p, v in zip( parser.types, parser.prefixes, parser.values, strict=True - )] + )) + return DnsForm(records=records, submit_value=parser.submit_value) + + +def parse_records(html: str) -> list[DnsRecord]: + """Compatibility helper for callers that only need the record list.""" + return list(parse_dns_form(html).records) + + +def normalize_cname_prefix(prefix: str, zone: str) -> str: + value = prefix.strip().rstrip(".").lower() + zone = zone.lower().rstrip(".") + if value.endswith("." + zone): + value = value[: -(len(zone) + 1)] + if not value or value == "@": + raise StratoError("CNAME prefix must name a subdomain, not the zone apex") + labels = value.split(".") + for index, label in enumerate(labels): + if label == "*": + if index != 0: + raise StratoError("A wildcard is allowed only in the first DNS label") + continue + ascii_label = label.encode("idna").decode("ascii") + if not re.fullmatch(r"[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?", ascii_label): + raise StratoError("CNAME prefix is not a valid relative DNS name") + return ".".join(label.encode("idna").decode("ascii") for label in labels) + + +def normalize_cname_target(target: str) -> str: + value = target.strip().rstrip(".").lower() + if not value or "://" in value or "/" in value: + raise StratoError("CNAME target must be a DNS name without scheme or path") + try: + ascii_value = value.encode("idna").decode("ascii") + except UnicodeError as exc: + raise StratoError("CNAME target is not a valid DNS name") from exc + if not re.fullmatch( + r"(?=.{1,253}$)[a-z0-9](?:[a-z0-9.-]*[a-z0-9])?", ascii_value + ) or any(not label or len(label) > 63 for label in ascii_value.split(".")): + raise StratoError("CNAME target is not a valid DNS name") + return ascii_value def parse_package_id(html: str, domain: str) -> str: @@ -300,6 +357,9 @@ class StratoClient: return self.package_id def list_txt_and_cname_records(self) -> list[DnsRecord]: + return list(self._get_dns_form().records) + + def _get_dns_form(self) -> DnsForm: if not self.session_id: self.login() package_id = self.resolve_package() @@ -313,7 +373,107 @@ class StratoClient: "vhost": self.config.domain, }, ) - return parse_records(html) + return parse_dns_form(html) def list_cnames(self) -> list[DnsRecord]: return [record for record in self.list_txt_and_cname_records() if record.type == "CNAME"] + + def _require_writes(self) -> None: + if not self.config.write_enabled: + raise StratoWriteDisabledError( + "DNS writes are disabled; set STRATO_WRITE_ENABLED=true to enable them" + ) + + def _push_records(self, form: DnsForm, records: list[DnsRecord]) -> None: + self._require_writes() + package_id = self.resolve_package() + self._request( + "POST", + form={ + "sessionID": self.session_id or "", + "cID": package_id, + "node": "ManageDomains", + "vhost": self.config.domain, + "prefix": [record.prefix for record in records], + "type": [record.type for record in records], + "value": [record.value for record in records], + "action_change_txt_records": form.submit_value, + }, + ) + + @staticmethod + def _cname_at(records: Iterable[DnsRecord], prefix: str) -> list[DnsRecord]: + return [ + record for record in records + if record.type == "CNAME" and record.prefix.lower() == prefix.lower() + ] + + def create_cname(self, prefix: str, target: str) -> DnsRecord: + normalized_prefix = normalize_cname_prefix(prefix, self.config.domain) + normalized_target = normalize_cname_target(target) + form = self._get_dns_form() + collisions = [ + record for record in form.records + if record.prefix.lower() == normalized_prefix.lower() + ] + if collisions: + raise StratoError("A DNS record with this prefix already exists") + created = DnsRecord("CNAME", normalized_prefix, normalized_target) + self._push_records(form, [*form.records, created]) + verified = self._cname_at(self._get_dns_form().records, normalized_prefix) + if len(verified) != 1 or normalize_cname_target(verified[0].value) != normalized_target: + raise StratoError("STRATO did not persist the new CNAME record") + return verified[0] + + def update_cname( + self, + prefix: str, + target: str, + *, + new_prefix: str | None = None, + ) -> tuple[DnsRecord, DnsRecord]: + old_prefix = normalize_cname_prefix(prefix, self.config.domain) + destination_prefix = normalize_cname_prefix( + new_prefix if new_prefix is not None else prefix, + self.config.domain, + ) + normalized_target = normalize_cname_target(target) + form = self._get_dns_form() + matches = self._cname_at(form.records, old_prefix) + if len(matches) != 1: + raise StratoError("Exactly one existing CNAME with this prefix is required") + if destination_prefix.lower() != old_prefix.lower() and any( + record.prefix.lower() == destination_prefix.lower() for record in form.records + ): + raise StratoError("A DNS record with the new prefix already exists") + replacement = DnsRecord("CNAME", destination_prefix, normalized_target) + updated_records = [ + replacement if record is matches[0] else record for record in form.records + ] + self._push_records(form, updated_records) + verified_form = self._get_dns_form() + verified = self._cname_at(verified_form.records, destination_prefix) + old_remaining = ( + self._cname_at(verified_form.records, old_prefix) + if destination_prefix.lower() != old_prefix.lower() + else [] + ) + if ( + len(verified) != 1 + or old_remaining + or normalize_cname_target(verified[0].value) != normalized_target + ): + raise StratoError("STRATO did not persist the CNAME update") + return matches[0], verified[0] + + def delete_cname(self, prefix: str) -> DnsRecord: + normalized_prefix = normalize_cname_prefix(prefix, self.config.domain) + form = self._get_dns_form() + matches = self._cname_at(form.records, normalized_prefix) + if len(matches) != 1: + raise StratoError("Exactly one existing CNAME with this prefix is required") + remaining = [record for record in form.records if record is not matches[0]] + self._push_records(form, remaining) + if self._cname_at(self._get_dns_form().records, normalized_prefix): + raise StratoError("STRATO did not delete the CNAME record") + return matches[0] diff --git a/strato_mcp.py b/strato_mcp.py index 23bd6ba..04f77d0 100644 --- a/strato_mcp.py +++ b/strato_mcp.py @@ -1,5 +1,5 @@ #!/usr/bin/env python3 -"""Read-only STRATO DNS MCP proof of concept.""" +"""STRATO DNS MCP for a single configured zone.""" from __future__ import annotations @@ -12,10 +12,10 @@ from strato_client import StratoClient, StratoConfig, StratoError mcp = FastMCP( - "strato-dns-readonly", + "strato-dns", instructions=( - "Read the configured STRATO DNS zone. This experimental server is " - "strictly read-only and cannot create, change, or delete DNS records." + "Read and manage CNAME records in the configured STRATO DNS zone. " + "Write tools are available only when STRATO_WRITE_ENABLED=true." ), host="0.0.0.0", port=int(os.environ.get("PORT", "8000")), @@ -39,6 +39,7 @@ def strato_connection_status() -> str: "record_count": len(records), "cname_count": sum(record.type == "CNAME" for record in records), "read_only": True, + "write_enabled": config.write_enabled, }) except StratoError as exc: return _json({"connected": False, "error": str(exc), "read_only": True}) @@ -56,10 +57,55 @@ def strato_list_cnames() -> str: "records": [record.as_dict() for record in records[:200]], "truncated": len(records) > 200, "read_only": True, + "write_enabled": config.write_enabled, }) except StratoError as exc: return _json({"error": str(exc), "read_only": True}) +@mcp.tool() +def strato_create_cname(prefix: str, target: str) -> str: + """Create one CNAME. Prefix may be relative or end in the configured zone.""" + try: + config = StratoConfig.from_env() + record = StratoClient(config).create_cname(prefix, target) + return _json({"created": True, "domain": config.domain, "record": record.as_dict()}) + except StratoError as exc: + return _json({"created": False, "error": str(exc)}) + + +@mcp.tool() +def strato_update_cname(prefix: str, target: str, new_prefix: str | None = None) -> str: + """Change a CNAME target and optionally rename its prefix.""" + try: + config = StratoConfig.from_env() + before, after = StratoClient(config).update_cname( + prefix, target, new_prefix=new_prefix + ) + return _json({ + "updated": True, + "domain": config.domain, + "before": before.as_dict(), + "after": after.as_dict(), + }) + except StratoError as exc: + return _json({"updated": False, "error": str(exc)}) + + +@mcp.tool() +def strato_delete_cname(prefix: str) -> str: + """Delete exactly one existing CNAME identified by its prefix.""" + try: + config = StratoConfig.from_env() + deleted = StratoClient(config).delete_cname(prefix) + return _json({ + "deleted": True, + "domain": config.domain, + "record": deleted.as_dict(), + }) + except StratoError as exc: + return _json({"deleted": False, "error": str(exc)}) + + if __name__ == "__main__": mcp.run(transport=os.environ.get("MCP_TRANSPORT", "streamable-http")) diff --git a/tests/test_strato_readonly.py b/tests/test_strato_readonly.py index 231e3bd..1947f36 100644 --- a/tests/test_strato_readonly.py +++ b/tests/test_strato_readonly.py @@ -5,6 +5,7 @@ import importlib.util import os import sys import unittest +import urllib.parse from pathlib import Path @@ -39,6 +40,19 @@ class FakeOpener: 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'' + f'' + f'' + ) + fields.append('') + return "".join(fields) + def test_dns_form_is_parsed_without_script_or_markup(self) -> None: html = """ @@ -47,6 +61,7 @@ class StratoParserTests(unittest.TestCase): + """ records = module.parse_records(html) self.assertEqual(records[0].as_dict(), { @@ -91,11 +106,7 @@ class StratoParserTests(unittest.TestCase):
example.de Hosting open
""" - records_html = """ - - - - """ + 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"), @@ -111,6 +122,133 @@ class StratoParserTests(unittest.TestCase): 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()