Add read-only STRATO DNS MCP prototype

services/strato-dns-mcp/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"]