Add read-only STRATO DNS MCP prototype

.gitignore

@@ -0,0 +1,7 @@
+__pycache__/
+*.py[cod]
+.pytest_cache/
+.venv/
+strato.env
+.env
+

Dockerfile

@@ -0,0 +1,22 @@
+FROM python:3.13-slim@sha256:ffb752e139c0a19692a43af8d8523b274222dd68eebad5d583b45c2201c6e30a
+
+ARG MCP_VERSION=1.29.0
+RUN pip install --no-cache-dir "mcp==${MCP_VERSION}" \
+    && groupadd --system --gid 10009 stratomcp \
+    && useradd --system --uid 10009 --gid 10009 --no-create-home stratomcp
+
+COPY strato_client.py /app/strato_client.py
+COPY strato_mcp.py /app/strato_mcp.py
+
+USER 10009:10009
+WORKDIR /app
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    MCP_TRANSPORT=streamable-http \
+    PORT=8000
+EXPOSE 8000
+
+HEALTHCHECK --interval=30s --timeout=5s --start-period=15s --retries=3 \
+  CMD python -c "import socket; s=socket.create_connection(('127.0.0.1',8000),3); s.close()"
+
+ENTRYPOINT ["python", "/app/strato_mcp.py"]

README.md

@@ -0,0 +1,57 @@
+# STRATO-DNS-MCP – Read-only-Prototyp
+
+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.
+
+## 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.
+
+Referenzstand der Untersuchung:
+`FlixMa/certbot-dns-strato@67df6dcfc3ef0035ec5aa5daf7c5b0bd8310d7fb`.
+
+## Lokaler Test – noch nicht produktiv installieren
+
+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.
+
+Das Docker-Image wird direkt aus diesem Repository gebaut:
+
+```sh
+docker build -t strato-dns-mcp:readonly .
+```
+
+Die Offline-Tests benötigen keine Zugangsdaten und kontaktieren STRATO nicht:
+
+```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 bewusst nicht enthalten
+
+- 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.

strato.env.example

@@ -0,0 +1,13 @@
+# Nur als lokale Vorlage. Niemals echte Werte in Git committen.
+STRATO_USERNAME=CHANGE_ME
+STRATO_PASSWORD=CHANGE_ME
+STRATO_DOMAIN=example.de
+
+# Optional: spart die Paket-Erkennung, falls die cID bekannt ist.
+STRATO_PACKAGE_ID=
+
+# Nur bei aktiviertem STRATO-TOTP notwendig. Diese Werte bleiben lokal.
+STRATO_TOTP_SECRET=
+STRATO_TOTP_DEVICE=
+
+STRATO_TIMEOUT_SECONDS=20

strato_client.py

@@ -0,0 +1,319 @@
+#!/usr/bin/env python3
+"""Small read-only 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.
+"""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import hmac
+import os
+import re
+import struct
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+from dataclasses import dataclass
+from html.parser import HTMLParser
+from http.cookiejar import CookieJar
+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)"
+
+
+class StratoError(RuntimeError):
+    """Short credential-free error suitable for an MCP response."""
+
+
+class StratoAuthenticationError(StratoError):
+    pass
+
+
+class StratoParseError(StratoError):
+    pass
+
+
+@dataclass(frozen=True)
+class StratoConfig:
+    username: str
+    password: str
+    domain: str
+    package_id: str | None = None
+    totp_secret: str | None = None
+    totp_device: str | None = None
+    timeout_seconds: float = 20.0
+
+    @classmethod
+    def from_env(cls) -> "StratoConfig":
+        values = {
+            "username": os.environ.get("STRATO_USERNAME", "").strip(),
+            "password": os.environ.get("STRATO_PASSWORD", ""),
+            "domain": os.environ.get("STRATO_DOMAIN", "").strip().rstrip("."),
+        }
+        missing = [name.upper() for name, value in values.items() if not value]
+        if missing:
+            raise StratoError(
+                "Missing configuration: " + ", ".join(f"STRATO_{name}" for name in missing)
+            )
+        domain = values["domain"].encode("idna").decode("ascii").lower()
+        if not re.fullmatch(r"(?=.{1,253}$)[a-z0-9](?:[a-z0-9.-]*[a-z0-9])?", domain):
+            raise StratoError("STRATO_DOMAIN is not a valid DNS zone name")
+        return cls(
+            username=values["username"],
+            password=values["password"],
+            domain=domain,
+            package_id=os.environ.get("STRATO_PACKAGE_ID", "").strip() or None,
+            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")),
+        )
+
+
+@dataclass(frozen=True)
+class DnsRecord:
+    type: str
+    prefix: str
+    value: str
+
+    def as_dict(self) -> dict[str, str]:
+        return {"type": self.type, "prefix": self.prefix, "value": self.value}
+
+
+class _FormParser(HTMLParser):
+    """Extract parallel type/prefix/value fields from STRATO's DNS form."""
+
+    def __init__(self) -> None:
+        super().__init__(convert_charrefs=True)
+        self.prefixes: list[str] = []
+        self.types: list[str] = []
+        self.values: list[str] = []
+        self._in_type_select = False
+        self._selected_option = False
+        self._option_value = ""
+        self._in_value_textarea = False
+        self._textarea_parts: list[str] = []
+
+    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 == "select" and data.get("name") == "type":
+            self._in_type_select = True
+        elif tag == "option" and self._in_type_select:
+            self._selected_option = "selected" in data
+            self._option_value = data.get("value") or ""
+            if self._selected_option:
+                self.types.append(self._option_value)
+        elif tag == "textarea" and data.get("name") == "value":
+            self._in_value_textarea = True
+            self._textarea_parts = []
+
+    def handle_endtag(self, tag: str) -> None:
+        if tag == "select":
+            self._in_type_select = False
+        elif tag == "option":
+            self._selected_option = False
+        elif tag == "textarea" and self._in_value_textarea:
+            self.values.append("".join(self._textarea_parts))
+            self._in_value_textarea = False
+
+    def handle_data(self, data: str) -> None:
+        if self._in_value_textarea:
+            self._textarea_parts.append(data)
+
+
+class _PackageParser(HTMLParser):
+    def __init__(self) -> None:
+        super().__init__(convert_charrefs=True)
+        self.rows: list[tuple[str, list[str]]] = []
+        self._depth = 0
+        self._text: list[str] = []
+        self._links: list[str] = []
+
+    def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
+        if tag == "tr":
+            if self._depth == 0:
+                self._text, self._links = [], []
+            self._depth += 1
+        if self._depth and tag == "a":
+            href = dict(attrs).get("href")
+            if href:
+                self._links.append(href)
+
+    def handle_endtag(self, tag: str) -> None:
+        if tag == "tr" and self._depth:
+            self._depth -= 1
+            if self._depth == 0:
+                self.rows.append((" ".join(self._text), list(self._links)))
+
+    def handle_data(self, data: str) -> None:
+        if self._depth and data.strip():
+            self._text.append(data.strip())
+
+
+def _totp(secret: str, at_time: int | None = None) -> str:
+    """Generate a standard six-digit SHA-1 TOTP without third-party modules."""
+    normalized = re.sub(r"\s+", "", secret).upper()
+    try:
+        key = base64.b32decode(normalized + "=" * ((8 - len(normalized) % 8) % 8))
+    except Exception as exc:
+        raise StratoAuthenticationError("STRATO_TOTP_SECRET is not valid base32") from exc
+    counter = int((at_time if at_time is not None else time.time()) // 30)
+    digest = hmac.new(key, struct.pack(">Q", counter), hashlib.sha1).digest()
+    offset = digest[-1] & 0x0F
+    number = struct.unpack(">I", digest[offset : offset + 4])[0] & 0x7FFFFFFF
+    return f"{number % 1_000_000:06d}"
+
+
+def parse_records(html: str) -> list[DnsRecord]:
+    parser = _FormParser()
+    parser.feed(html)
+    counts = (len(parser.types), len(parser.prefixes), len(parser.values))
+    if len(set(counts)) != 1:
+        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(
+        parser.types, parser.prefixes, parser.values, strict=True
+    )]
+
+
+def parse_package_id(html: str, domain: str) -> str:
+    parser = _PackageParser()
+    parser.feed(html)
+    for text, links in parser.rows:
+        if domain.lower() not in text.lower():
+            continue
+        for link in links:
+            package = urllib.parse.parse_qs(urllib.parse.urlparse(link).query).get("cID")
+            if package and package[0].isdigit():
+                return package[0]
+    raise StratoParseError(f"Configured domain {domain} was not found in STRATO packages")
+
+
+class StratoClient:
+    def __init__(
+        self,
+        config: StratoConfig,
+        *,
+        opener: object | None = None,
+        sleep: Callable[[float], None] = time.sleep,
+    ) -> None:
+        self.config = config
+        self.opener = opener or urllib.request.build_opener(
+            urllib.request.HTTPCookieProcessor(CookieJar())
+        )
+        self.sleep = sleep
+        self.session_id: str | None = None
+        self.package_id: str | None = config.package_id
+
+    def _request(
+        self,
+        method: str,
+        *,
+        params: dict[str, object] | None = None,
+        form: dict[str, object] | None = None,
+    ) -> tuple[str, str]:
+        url = STRATO_URL
+        if params:
+            url += "?" + urllib.parse.urlencode(params, doseq=True)
+        body = urllib.parse.urlencode(form, doseq=True).encode() if form is not None else None
+        request = urllib.request.Request(
+            url,
+            data=body,
+            method=method,
+            headers={"User-Agent": USER_AGENT, "Accept": "text/html,application/xhtml+xml"},
+        )
+        try:
+            response = self.opener.open(request, timeout=self.config.timeout_seconds)
+            raw = response.read(MAX_RESPONSE_BYTES + 1)
+        except urllib.error.HTTPError as exc:
+            raise StratoError(f"STRATO returned HTTP {exc.code}") from None
+        except (urllib.error.URLError, TimeoutError, OSError):
+            raise StratoError("STRATO could not be reached") from None
+        if len(raw) > MAX_RESPONSE_BYTES:
+            raise StratoError("STRATO response exceeded the size limit")
+        return response.geturl(), raw.decode("utf-8", errors="replace")
+
+    def login(self) -> None:
+        self._request("GET")
+        self.sleep(1.0)
+        url, html = self._request(
+            "POST",
+            form={
+                "identifier": self.config.username,
+                "passwd": self.config.password,
+                "action_customer_login.x": "Login",
+            },
+        )
+        if re.search(r"Zwei.Faktor.Authentifizierung", html, flags=re.IGNORECASE):
+            if not self.config.totp_secret or not self.config.totp_device:
+                raise StratoAuthenticationError(
+                    "STRATO requested 2FA; configure STRATO_TOTP_SECRET and STRATO_TOTP_DEVICE"
+                )
+            token = re.search(r'name=["\']totp_token["\'][^>]*value=["\']([^"\']+)', html)
+            device = re.search(
+                rf'<option\s+value=["\'](S\.{re.escape(self.config.username)}\.\w+)["\'][^>]*>'
+                rf'\s*{re.escape(self.config.totp_device)}\s*</option>',
+                html,
+                flags=re.IGNORECASE,
+            )
+            if not token or not device:
+                raise StratoParseError("STRATO 2FA form could not be understood")
+            self.sleep(1.0)
+            url, html = self._request(
+                "POST",
+                form={
+                    "identifier": self.config.username,
+                    "totp_token": token.group(1),
+                    "pw_id": device.group(1),
+                    "totp": _totp(self.config.totp_secret),
+                    "action_customer_login.x": 1,
+                },
+            )
+        session = urllib.parse.parse_qs(urllib.parse.urlparse(url).query).get("sessionID")
+        if not session:
+            raise StratoAuthenticationError("STRATO login was not accepted")
+        self.session_id = session[0]
+
+    def resolve_package(self) -> str:
+        if self.package_id:
+            return self.package_id
+        if not self.session_id:
+            raise StratoAuthenticationError("STRATO session is not initialized")
+        _, html = self._request(
+            "GET",
+            params={"sessionID": self.session_id, "cID": 0, "node": "kds_CustomerEntryPage"},
+        )
+        self.package_id = parse_package_id(html, self.config.domain)
+        return self.package_id
+
+    def list_txt_and_cname_records(self) -> list[DnsRecord]:
+        if not self.session_id:
+            self.login()
+        package_id = self.resolve_package()
+        _, html = self._request(
+            "GET",
+            params={
+                "sessionID": self.session_id or "",
+                "cID": package_id,
+                "node": "ManageDomains",
+                "action_show_txt_records": "",
+                "vhost": self.config.domain,
+            },
+        )
+        return parse_records(html)
+
+    def list_cnames(self) -> list[DnsRecord]:
+        return [record for record in self.list_txt_and_cname_records() if record.type == "CNAME"]

strato_mcp.py

@@ -0,0 +1,65 @@
+#!/usr/bin/env python3
+"""Read-only STRATO DNS MCP proof of concept."""
+
+from __future__ import annotations
+
+import json
+import os
+
+from mcp.server.fastmcp import FastMCP
+
+from strato_client import StratoClient, StratoConfig, StratoError
+
+
+mcp = FastMCP(
+    "strato-dns-readonly",
+    instructions=(
+        "Read the configured STRATO DNS zone. This experimental server is "
+        "strictly read-only and cannot create, change, or delete DNS records."
+    ),
+    host="0.0.0.0",
+    port=int(os.environ.get("PORT", "8000")),
+    stateless_http=True,
+)
+
+
+def _json(value: object) -> str:
+    return json.dumps(value, ensure_ascii=False, separators=(",", ":"))
+
+
+@mcp.tool()
+def strato_connection_status() -> str:
+    """Log in and verify that the configured DNS zone can be read. Makes no change."""
+    try:
+        config = StratoConfig.from_env()
+        records = StratoClient(config).list_txt_and_cname_records()
+        return _json({
+            "connected": True,
+            "domain": config.domain,
+            "record_count": len(records),
+            "cname_count": sum(record.type == "CNAME" for record in records),
+            "read_only": True,
+        })
+    except StratoError as exc:
+        return _json({"connected": False, "error": str(exc), "read_only": True})
+
+
+@mcp.tool()
+def strato_list_cnames() -> str:
+    """List CNAME records for the one configured STRATO zone. Makes no change."""
+    try:
+        config = StratoConfig.from_env()
+        records = StratoClient(config).list_cnames()
+        return _json({
+            "domain": config.domain,
+            "count": len(records),
+            "records": [record.as_dict() for record in records[:200]],
+            "truncated": len(records) > 200,
+            "read_only": True,
+        })
+    except StratoError as exc:
+        return _json({"error": str(exc), "read_only": True})
+
+
+if __name__ == "__main__":
+    mcp.run(transport=os.environ.get("MCP_TRANSPORT", "streamable-http"))

tests/test_strato_readonly.py

@@ -0,0 +1,116 @@
+#!/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] / "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&amp;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&amp;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()