480 lines
18 KiB
Python
480 lines
18 KiB
Python
#!/usr/bin/env python3
|
|
"""Small HTTP client for STRATO's customer portal.
|
|
|
|
STRATO does not publish a DNS-zone API for ordinary hosted domains. This
|
|
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
|
|
|
|
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-mcp/0.2)"
|
|
|
|
|
|
class StratoError(RuntimeError):
|
|
"""Short credential-free error suitable for an MCP response."""
|
|
|
|
|
|
class StratoAuthenticationError(StratoError):
|
|
pass
|
|
|
|
|
|
class StratoParseError(StratoError):
|
|
pass
|
|
|
|
|
|
class StratoWriteDisabledError(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
|
|
write_enabled: bool = False
|
|
|
|
@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")),
|
|
write_enabled=os.environ.get("STRATO_WRITE_ENABLED", "false").strip().lower()
|
|
in {"1", "true", "yes", "on"},
|
|
)
|
|
|
|
|
|
@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}
|
|
|
|
|
|
@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."""
|
|
|
|
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] = []
|
|
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:
|
|
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_dns_form(html: str) -> DnsForm:
|
|
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"
|
|
)
|
|
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:
|
|
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]:
|
|
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()
|
|
_, 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_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]
|