commit 4754626d34648f69832ece8dd4dad0e9436687d1 Author: Mikei386 <44135113+Mikei386@users.noreply.github.com> Date: Fri Aug 28 19:55:20 2026 +0200 Add read-only STRATO DNS MCP prototype diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ed5f341 --- /dev/null +++ b/.gitignore @@ -0,0 +1,7 @@ +__pycache__/ +*.py[cod] +.pytest_cache/ +.venv/ +strato.env +.env + diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..4729491 --- /dev/null +++ b/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"] diff --git a/README.md b/README.md new file mode 100644 index 0000000..40ffe73 --- /dev/null +++ b/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. diff --git a/strato.env.example b/strato.env.example new file mode 100644 index 0000000..9e3ef99 --- /dev/null +++ b/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 diff --git a/strato_client.py b/strato_client.py new file mode 100644 index 0000000..87ee00f --- /dev/null +++ b/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']*>' + rf'\s*{re.escape(self.config.totp_device)}\s*', + 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"] diff --git a/strato_mcp.py b/strato_mcp.py new file mode 100644 index 0000000..23bd6ba --- /dev/null +++ b/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")) diff --git a/tests/test_strato_readonly.py b/tests/test_strato_readonly.py new file mode 100644 index 0000000..231e3bd --- /dev/null +++ b/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 = """ + + + + + + + """ + 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('') + + def test_package_is_selected_by_domain_not_fallback(self) -> None: + html = """ + +
other.exampleopen
example.de Hostingopen
+ """ + 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 = """ + +
example.de Hostingopen
+ """ + records_html = """ + + + + """ + 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()