fix: harden live cpanel mutation paths

Author: MNametissa

.gitignore

@@ -1,9 +1,9 @@
 .venv/
+.venv-dev/
 __pycache__/
 .pytest_cache/
 .ruff_cache/
 .mypy_cache/
 .coverage
 *.egg-info/
 .env
-

o2switch_cli/cli/helpers.py

@@ -13,19 +13,36 @@
 T = TypeVar("T")
 
 
-def run_guarded(ctx: typer.Context, action: Callable[[AppContext], T]) -> T:
-    app_context = get_app_context(ctx)
+def _execute_guarded(app_context: AppContext, action: Callable[[AppContext], T]) -> T:
     try:
         return action(app_context)
     except CliAppError as error:
-        raise_for_error(app_context, error)
+        raise error
     except typer.Exit:
         raise
     except Exception as error:  # pragma: no cover
-        raise_for_error(app_context, TransportAppError("runtime", str(error)))
+        raise TransportAppError("runtime", str(error)) from error
     raise RuntimeError("unreachable")
 
 
+def run_guarded(ctx: typer.Context, action: Callable[[AppContext], T]) -> T:
+    app_context = get_app_context(ctx)
+    try:
+        return _execute_guarded(app_context, action)
+    except CliAppError as error:
+        raise_for_error(app_context, error)
+    raise RuntimeError("unreachable")
+
+
+def run_guarded_interactive(app_context: AppContext, action: Callable[[AppContext], T]) -> T | None:
+    ui = TerminalUI(app_context.console, app_context.output_format)
+    try:
+        return _execute_guarded(app_context, action)
+    except CliAppError as error:
+        ui.print_error(error.to_envelope())
+        return None
+
+
 def confirm_plan(app_context: AppContext, ui: TerminalUI, plan: MutationPlan, *, zone: str | None = None) -> bool:
     if ui.output_format != "json":
         ui.print_plan(plan, zone=zone)

o2switch_cli/cli/interactive.py

@@ -9,6 +9,7 @@
 from o2switch_cli.cli.context import build_context
 from o2switch_cli.cli.dns import app as dns_app
 from o2switch_cli.cli.domains import app as domains_app
+from o2switch_cli.cli.helpers import run_guarded
 from o2switch_cli.cli.interactive import run_interactive_menu
 from o2switch_cli.cli.subdomains import app as subdomains_app
 from o2switch_cli.config.settings import load_settings
@@ -52,16 +53,15 @@ def main(
 
     if ctx.invoked_subcommand is None and not ctx.resilient_parsing:
         if sys.stdin.isatty():
-            run_interactive_menu(app_context)
+            run_guarded(ctx, lambda active_context: run_interactive_menu(active_context))
             raise typer.Exit()
         typer.echo(ctx.get_help())
         raise typer.Exit()
 
 
 @app.command("interactive")
 def interactive(ctx: typer.Context) -> None:
-    app_context = ctx.obj
-    run_interactive_menu(app_context)
+    run_guarded(ctx, run_interactive_menu)
 
 
 def run() -> None:

o2switch_cli/cli/main.py

@@ -71,13 +71,34 @@ def _parse_api2(self, payload: dict[str, Any], operation: str) -> ApiResult:
         event = result.get("event", {})
         if event and event.get("result") not in (1, "1"):
             raise TransportAppError(operation, event.get("reason", f"{operation} failed."))
+        data = result.get("data")
+        if isinstance(data, dict):
+            detail = self._api2_failure_detail(data, operation)
+            if detail:
+                raise TransportAppError(operation, detail)
+        elif isinstance(data, list):
+            for item in data:
+                if not isinstance(item, dict):
+                    continue
+                detail = self._api2_failure_detail(item, operation)
+                if detail:
+                    raise TransportAppError(operation, detail)
         return ApiResult(
-            data=result.get("data"),
+            data=data,
             metadata=result.get("metadata", {}) or {},
             warnings=result.get("warnings", []) or [],
             messages=result.get("messages", []) or [],
         )
 
+    @staticmethod
+    def _api2_failure_detail(payload: dict[str, Any], operation: str) -> str | None:
+        if "result" not in payload:
+            return None
+        if payload.get("result") in (1, "1", True):
+            return None
+        detail = payload.get("reason") or payload.get("statusmsg") or payload.get("error") or f"{operation} failed."
+        return str(detail)
+
     def uapi(self, module: str, function: str, **params: Any) -> ApiResult:
         payload = self._request("GET", f"/execute/{module}/{function}", params=params)
         return self._parse_uapi(payload, f"{module}/{function}")
@@ -128,8 +149,8 @@ def add_subdomain(self, *, domain: str, rootdomain: str, directory: str) -> ApiR
     def list_subdomains(self) -> ApiResult:
         return self.api2("SubDomain", "listsubdomains")
 
-    def delete_subdomain(self, *, domain: str, rootdomain: str) -> ApiResult:
-        return self.api2("SubDomain", "delsubdomain", domain=domain, rootdomain=rootdomain)
+    def delete_subdomain(self, *, domain: str) -> ApiResult:
+        return self.api2("SubDomain", "delsubdomain", domain=domain)
 
     def test_access(self) -> ApiResult:
         return self.list_domains()

o2switch_cli/core/cpanel_client.py

@@ -1,5 +1,8 @@
 from __future__ import annotations
 
+import base64
+import binascii
+
 from o2switch_cli.core.audit import AuditService
 from o2switch_cli.core.cpanel_client import CpanelClient
 from o2switch_cli.core.domain_service import DomainService
@@ -45,6 +48,50 @@ def _line_index(record: DNSRecord) -> int:
         raw_value = record.record_id or record.raw.get("line_index") or record.raw.get("line")
         return int(raw_value)
 
+    @staticmethod
+    def _decode_b64_text(value: object) -> str | None:
+        if not isinstance(value, str) or not value:
+            return None
+        try:
+            decoded = base64.b64decode(value, validate=True)
+        except (binascii.Error, ValueError):
+            return None
+        try:
+            return decoded.decode("utf-8")
+        except UnicodeDecodeError:
+            return decoded.decode("utf-8", errors="replace")
+
+    @classmethod
+    def _record_values(cls, item: dict[str, object]) -> list[str]:
+        data = item.get("data")
+        if isinstance(data, list):
+            return [str(value) for value in data]
+        if data not in (None, ""):
+            return [str(data)]
+
+        data_b64 = item.get("data_b64")
+        if isinstance(data_b64, list):
+            return [decoded for value in data_b64 if (decoded := cls._decode_b64_text(value)) is not None]
+        if decoded := cls._decode_b64_text(data_b64):
+            return [decoded]
+        return []
+
+    @classmethod
+    def _record_name(cls, item: dict[str, object], root_domain: str) -> str:
+        raw_name = item.get("dname") or item.get("name") or item.get("domain")
+        if raw_name in (None, ""):
+            raw_name = cls._decode_b64_text(item.get("dname_b64")) or root_domain
+        return canonical_record_name(str(raw_name), root_domain)
+
+    @staticmethod
+    def _require_serial(operation: str, root_domain: str, serial: int | None) -> int:
+        if serial is None:
+            raise TransportAppError(
+                operation,
+                f"Could not determine the current DNS serial for zone {root_domain}.",
+            )
+        return serial
+
     def _zone_state(self, root_domain: str) -> tuple[list[DNSRecord], int | None]:
         result = self._client.parse_zone(root_domain)
         payload = result.data or {}
@@ -64,13 +111,7 @@ def _zone_state(self, root_domain: str) -> tuple[list[DNSRecord], int | None]:
             if not isinstance(item, dict):
                 continue
             record_type = str(item.get("record_type") or item.get("type") or "").upper()
-            data = item.get("data")
-            if isinstance(data, list):
-                values = [str(value) for value in data]
-            elif data is None:
-                values = []
-            else:
-                values = [str(data)]
+            values = self._record_values(item)
             value = (
                 item.get("address")
                 or item.get("target")
@@ -80,10 +121,7 @@ def _zone_state(self, root_domain: str) -> tuple[list[DNSRecord], int | None]:
             )
             ttl = item.get("ttl")
             record_id = item.get("line_index") or item.get("line") or item.get("record_id") or item.get("id")
-            name = canonical_record_name(
-                str(item.get("dname") or item.get("name") or item.get("domain") or root_domain),
-                root_domain,
-            )
+            name = self._record_name(item, root_domain)
             records.append(
                 DNSRecord(
                     name=name,
@@ -106,15 +144,10 @@ def _extract_serial(records: list[DNSRecord]) -> int | None:
         for record in records:
             if record.type != "SOA":
                 continue
-            values = record.raw.get("data", [])
-            if isinstance(values, list):
-                for value in values:
-                    if str(value).isdigit() and len(str(value)) >= 8:
-                        return int(value)
-            elif isinstance(values, str):
-                for part in values.split():
-                    if part.isdigit() and len(part) >= 8:
-                        return int(part)
+            values = DNSService._record_values(record.raw)
+            for value in values:
+                if str(value).isdigit() and len(str(value)) >= 8:
+                    return int(value)
         return None
 
     def get_zone_state(self, root_domain: str) -> list[DNSRecord]:
@@ -318,6 +351,7 @@ def upsert_a_record(
         applied = False
         action = "dry-run" if dry_run else "created"
         if not dry_run:
+            serial = self._require_serial("dns_upsert", root_domain, serial)
             add = None
             edit = None
             remove = None
@@ -407,6 +441,7 @@ def delete_a_record(
         hostname = normalize_hostname(fqdn)
         root_domain, serial, matches, plan = self.plan_delete_a_record(hostname, force=force)
         if not dry_run:
+            serial = self._require_serial("dns_delete", root_domain, serial)
             remove = [self._line_index(record) for record in matches]
             self._client.mass_edit_zone(zone=root_domain, serial=serial, remove=remove)
         verification = VerificationStatus.SKIPPED

o2switch_cli/core/dns_service.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from o2switch_cli.core.cpanel_client import CpanelClient
+from o2switch_cli.core.errors import NotFoundAppError
 from o2switch_cli.core.models import DomainDescriptor, DomainType
 from o2switch_cli.core.validators import normalize_hostname, select_root_domain
 
@@ -48,5 +49,12 @@ def search(self, term: str) -> list[DomainDescriptor]:
         needle = term.strip().lower()
         return [item for item in self.list_domains() if needle in item.domain]
 
+    def get_domain_descriptor(self, domain: str, operation: str) -> DomainDescriptor:
+        target = normalize_hostname(domain)
+        for item in self.list_domains():
+            if item.domain == target:
+                return item
+        raise NotFoundAppError(operation, "Root domain was not found on the account.", target)
+
     def resolve_root_domain(self, hostname: str, operation: str) -> str:
         return select_root_domain(hostname, self.root_domains(), operation)

o2switch_cli/core/domain_service.py

@@ -11,6 +11,7 @@
     TransportAppError,
 )
 from o2switch_cli.core.models import (
+    DomainType,
     MutationPlan,
     OperationMode,
     OperationResult,
@@ -40,6 +41,13 @@ def __init__(
         self._audit = audit
         self._reserved_labels = reserved_labels
 
+    def _delete_domain_argument(self, hostname: str, root_domain: str) -> str:
+        descriptor = self._domains.get_domain_descriptor(root_domain, "subdomain_delete")
+        if descriptor.type is DomainType.ADDON:
+            label = hostname[: -(len(root_domain) + 1)]
+            return f"{label}_{root_domain}"
+        return hostname
+
     def search(self, term: str) -> list[SubdomainDescriptor]:
         needle = term.strip().lower()
         try:
@@ -188,7 +196,8 @@ def delete(self, fqdn: str, *, dry_run: bool) -> OperationResult:
         root_domain, label, plan = self.plan_delete(hostname)
         if not dry_run:
             try:
-                self._client.delete_subdomain(domain=label, rootdomain=root_domain)
+                delete_domain = self._delete_domain_argument(hostname, root_domain)
+                self._client.delete_subdomain(domain=delete_domain)
             except TransportAppError as exc:
                 if self._looks_like_unsupported_delete(str(exc)):
                     raise NotSupportedAppError(

o2switch_cli/core/subdomain_service.py

@@ -111,9 +111,9 @@ def validate_reserved_hostname(hostname: str, root_domain: str, reserved_labels:
 
 
 def canonical_record_name(name: str, zone: str) -> str:
-    candidate = normalize_hostname(name)
+    candidate = name.strip().lower().rstrip(".")
     normalised_zone = normalize_hostname(zone)
-    if candidate == "@":
+    if not candidate or candidate == "@":
         return normalised_zone
     if candidate == normalised_zone or candidate.endswith(f".{normalised_zone}"):
         return candidate

o2switch_cli/core/validators.py

@@ -5,6 +5,7 @@
 from typer.testing import CliRunner
 
 from o2switch_cli.cli.main import app
+from o2switch_cli.core.errors import TransportAppError
 from o2switch_cli.core.models import (
     DomainDescriptor,
     DomainType,
@@ -221,3 +222,23 @@ class FakeRuntime:
     monkeypatch.setattr("o2switch_cli.cli.context.AppContext.runtime", lambda self: FakeRuntime())
     result = runner.invoke(app, ["--json", "dns", "verify", "--host", "odoo.ginutech.com"])
     assert result.exit_code == 7
+
+
+def test_interactive_command_renders_guarded_error(monkeypatch) -> None:
+    def boom(app_context):
+        raise TransportAppError("runtime", "interactive exploded")
+
+    monkeypatch.setattr("o2switch_cli.cli.main.run_interactive_menu", boom)
+    result = runner.invoke(
+        app,
+        ["--json", "interactive"],
+        env={
+            "O2SWITCH_CLI_CPANEL_HOST": "cpanel.example.test",
+            "O2SWITCH_CLI_CPANEL_USER": "demo",
+            "O2SWITCH_CLI_CPANEL_TOKEN": "secret-token",
+        },
+    )
+    payload = json.loads(result.output)
+    assert result.exit_code == 6
+    assert payload["error_class"] == "transport"
+    assert payload["message"] == "interactive exploded"