michael/Strato-MCP
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add CNAME create update and delete tools

Browse Source
This commit is contained in:
Mikei386
2026-08-28 21:28:45 +02:00
parent 4754626d34
commit c6ab0e48c1
5 changed files with 400 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
strato_client.py
+171 -11
View File
@@ -1,12 +1,11 @@
#!/usr/bin/env python3
"""Small read-only HTTP client for STRATO's customer portal.
"""Small 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.
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
@@ -29,7 +28,7 @@ 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)"
USER_AGENT = "Mozilla/5.0 (compatible; mike-ai-strato-dns-mcp/0.2)"
class StratoError(RuntimeError):
@@ -44,6 +43,10 @@ class StratoParseError(StratoError):
pass
class StratoWriteDisabledError(StratoError):
pass
@dataclass(frozen=True)
class StratoConfig:
username: str
@@ -53,6 +56,7 @@ class StratoConfig:
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":
@@ -77,6 +81,8 @@ class StratoConfig:
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"},
)
@@ -90,6 +96,12 @@ class DnsRecord:
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."""
@@ -103,11 +115,14 @@ class _FormParser(HTMLParser):
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:
@@ -176,7 +191,7 @@ def _totp(secret: str, at_time: int | None = None) -> str:
return f"{number % 1_000_000:06d}"
def parse_records(html: str) -> list[DnsRecord]:
def parse_dns_form(html: str) -> DnsForm:
parser = _FormParser()
parser.feed(html)
counts = (len(parser.types), len(parser.prefixes), len(parser.values))
@@ -184,9 +199,51 @@ def parse_records(html: str) -> list[DnsRecord]:
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(
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:
@@ -300,6 +357,9 @@ class StratoClient:
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()
@@ -313,7 +373,107 @@ class StratoClient:
"vhost": self.config.domain,
},
)
return parse_records(html)
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]