#!/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]