From da3571ef9f4d41abcaf841ee81d302ba2b14b865 Mon Sep 17 00:00:00 2001 From: Mikei386 <44135113+Mikei386@users.noreply.github.com> Date: Fri, 28 Aug 2026 19:56:04 +0200 Subject: [PATCH] Move STRATO MCP to standalone repository --- dev/test_strato_readonly.py | 116 -------- services/strato-dns-mcp/Dockerfile | 22 -- services/strato-dns-mcp/README.md | 45 --- services/strato-dns-mcp/strato.env.example | 13 - services/strato-dns-mcp/strato_client.py | 319 --------------------- services/strato-dns-mcp/strato_mcp.py | 65 ----- 6 files changed, 580 deletions(-) delete mode 100644 dev/test_strato_readonly.py delete mode 100644 services/strato-dns-mcp/Dockerfile delete mode 100644 services/strato-dns-mcp/README.md delete mode 100644 services/strato-dns-mcp/strato.env.example delete mode 100644 services/strato-dns-mcp/strato_client.py delete mode 100644 services/strato-dns-mcp/strato_mcp.py diff --git a/dev/test_strato_readonly.py b/dev/test_strato_readonly.py deleted file mode 100644 index bc36071..0000000 --- a/dev/test_strato_readonly.py +++ /dev/null @@ -1,116 +0,0 @@ -#!/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] / "services/strato-dns-mcp/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() diff --git a/services/strato-dns-mcp/Dockerfile b/services/strato-dns-mcp/Dockerfile deleted file mode 100644 index 68dd355..0000000 --- a/services/strato-dns-mcp/Dockerfile +++ /dev/null @@ -1,22 +0,0 @@ -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 services/strato-dns-mcp/strato_client.py /app/strato_client.py -COPY services/strato-dns-mcp/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/services/strato-dns-mcp/README.md b/services/strato-dns-mcp/README.md deleted file mode 100644 index 47ed1f0..0000000 --- a/services/strato-dns-mcp/README.md +++ /dev/null @@ -1,45 +0,0 @@ -# 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. - -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/services/strato-dns-mcp/strato.env.example b/services/strato-dns-mcp/strato.env.example deleted file mode 100644 index 9e3ef99..0000000 --- a/services/strato-dns-mcp/strato.env.example +++ /dev/null @@ -1,13 +0,0 @@ -# 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/services/strato-dns-mcp/strato_client.py b/services/strato-dns-mcp/strato_client.py deleted file mode 100644 index 87ee00f..0000000 --- a/services/strato-dns-mcp/strato_client.py +++ /dev/null @@ -1,319 +0,0 @@ -#!/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/services/strato-dns-mcp/strato_mcp.py b/services/strato-dns-mcp/strato_mcp.py deleted file mode 100644 index 23bd6ba..0000000 --- a/services/strato-dns-mcp/strato_mcp.py +++ /dev/null @@ -1,65 +0,0 @@ -#!/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"))