Add stock configuration backup and restore

src/nr2301/namespaces/maintenance.py

@@ -13,6 +13,8 @@
 
 
 _TIME_RE = re.compile(r"^([0-9]{1,2}):([0-9]{1,2})$")
+_CONFIG_RESTORE_CHUNK_BYTES = 1024 * 1024
+_CONFIG_RESTORE_MAX_BYTES = 200 * 1024 * 1024
 
 
 class TimedRebootSettings(TypedDict, total=False):
@@ -29,7 +31,9 @@ class MaintenanceRecoveryResult(TypedDict, total=False):
     action_response: dict[str, Any]
     boot_time_after: int
     boot_time_before: int
+    chunk_count: int
     outage_observed: bool
+    uploaded_bytes: int
 
 
 class MaintenanceNamespace:
@@ -92,6 +96,167 @@ def backup_config(
             )
         return response
 
+    def download_config_backup(
+        self,
+        *,
+        timeout: float | None = None,
+    ) -> bytes:
+        """Download the current stock-WebUI configuration backup bytes.
+
+        The returned blob is secret-bearing and may contain credentials. The
+        SDK intentionally does not log, parse or persist it.
+        """
+
+        payload = self._client.transport.file_download(
+            params={
+                "Action": "Download",
+                "file": "backup_config",
+                "dl": "1",
+            },
+            timeout=timeout,
+        )
+        if not payload:
+            raise ProtocolError("configuration backup download returned no data")
+        return payload
+
+    def restore_config_backup(
+        self,
+        backup: bytes | bytearray | memoryview,
+        *,
+        chunk_size: int = _CONFIG_RESTORE_CHUNK_BYTES,
+        action_timeout: float = 20.0,
+        recovery_attempts: int = 120,
+        recovery_delay: float = 1.0,
+        recovery_timeout: float = 3.0,
+        initial_delay: float = 1.0,
+    ) -> MaintenanceRecoveryResult:
+        """Restore a trusted configuration backup through the stock file CGI.
+
+        ACIY.3's frontend sends sequential raw `application/octet-stream`
+        POSTs to `/file.cgi?Action=Upload&file=restore_config`, using 1 MiB
+        chunks and no multipart wrapper or Content-Range header.
+
+        Configuration data is secret-bearing. The SDK never includes backup
+        bytes in diagnostics. A response containing `other error` is treated
+        as failure. The final upload is considered successful only after
+        management recovers and `boot_time` proves a new boot.
+        """
+
+        if not isinstance(backup, (bytes, bytearray, memoryview)):
+            raise TypeError("backup must be bytes-like")
+        payload = bytes(backup)
+        if not payload:
+            raise ValueError("backup must not be empty")
+        if len(payload) > _CONFIG_RESTORE_MAX_BYTES:
+            raise ValueError("backup exceeds the stock frontend 200 MiB limit")
+        if isinstance(chunk_size, bool) or not isinstance(chunk_size, int):
+            raise TypeError("chunk_size must be an int")
+        if chunk_size <= 0:
+            raise ValueError("chunk_size must be greater than zero")
+
+        self._validate_recovery_args(
+            action_timeout=action_timeout,
+            recovery_attempts=recovery_attempts,
+            recovery_delay=recovery_delay,
+            recovery_timeout=recovery_timeout,
+            initial_delay=initial_delay,
+        )
+        before = self._read_boot_time(timeout=recovery_timeout)
+        params = {"Action": "Upload", "file": "restore_config"}
+        chunk_count = (len(payload) + chunk_size - 1) // chunk_size
+        final_action_error: NR2301Error | None = None
+
+        for index in range(chunk_count):
+            start = index * chunk_size
+            end = min(len(payload), start + chunk_size)
+            chunk = payload[start:end]
+            final_chunk = index + 1 == chunk_count
+            try:
+                response = self._client.transport.file_upload(
+                    chunk,
+                    params=params,
+                    timeout=action_timeout,
+                )
+            except TransportError as exc:
+                if not final_chunk:
+                    raise APIError(
+                        "configuration restore upload failed before final chunk",
+                        method_id="file.cgi/restore_config",
+                        response={
+                            "chunk_index": index,
+                            "chunk_count": chunk_count,
+                            "uploaded_bytes_before_failure": start,
+                            "error": type(exc).__name__,
+                        },
+                    ) from exc
+                final_action_error = exc
+                break
+
+            if b"other error" in response.lower():
+                raise APIError(
+                    "configuration restore was rejected by file.cgi",
+                    method_id="file.cgi/restore_config",
+                    response={
+                        "chunk_index": index,
+                        "chunk_count": chunk_count,
+                        "uploaded_bytes_before_failure": start,
+                    },
+                )
+
+        started = time.monotonic()
+        if initial_delay:
+            time.sleep(initial_delay)
+
+        last_boot: int | None = None
+        last_error: NR2301Error | None = None
+        outage_observed = final_action_error is not None
+
+        for attempt in range(recovery_attempts):
+            try:
+                current = self._read_boot_time(timeout=recovery_timeout)
+                last_boot = current
+                elapsed = time.monotonic() - started
+                if current < before or (
+                    outage_observed
+                    and current <= int(elapsed) + 10
+                    and before <= int(elapsed) + 10
+                ):
+                    result: MaintenanceRecoveryResult = {
+                        "boot_time_before": before,
+                        "boot_time_after": current,
+                        "outage_observed": outage_observed,
+                        "uploaded_bytes": len(payload),
+                        "chunk_count": chunk_count,
+                    }
+                    if final_action_error is not None:
+                        result["action_error"] = type(final_action_error).__name__
+                    return result
+            except NR2301Error as exc:
+                outage_observed = True
+                last_error = exc
+                last_error = self._try_relogin(last_error)
+
+            if attempt + 1 < recovery_attempts and recovery_delay:
+                time.sleep(recovery_delay)
+
+        details: dict[str, Any] = {
+            "boot_time_before": before,
+            "boot_time_after": last_boot,
+            "outage_observed": outage_observed,
+            "uploaded_bytes": len(payload),
+            "chunk_count": chunk_count,
+        }
+        if final_action_error is not None:
+            details["action_error"] = type(final_action_error).__name__
+        if last_error is not None:
+            details["last_recovery_error"] = type(last_error).__name__
+
+        raise APIError(
+            "configuration restore could not be verified by recovery and boot_time reset",
+            method_id="file.cgi/restore_config",
+            response=details,
+        )
+
     def restart_web_server(
         self,
         *,

src/nr2301/transport.py

@@ -73,6 +73,49 @@ def multicall(
         )
         return self._decode_json(response, method_id="multicall")
 
+    @property
+    def file_url(self) -> str:
+        return f"{self.base_url}/file.cgi"
+
+    def file_download(
+        self,
+        *,
+        params: Mapping[str, Any],
+        timeout: float | None = None,
+    ) -> bytes:
+        """Download opaque bytes from the authenticated `/file.cgi` family."""
+
+        effective_timeout = self.timeout if timeout is None else timeout
+        response = self._request(
+            "GET",
+            self.file_url,
+            params=dict(params),
+            timeout=effective_timeout,
+        )
+        return bytes(response.content)
+
+    def file_upload(
+        self,
+        data: bytes,
+        *,
+        params: Mapping[str, Any],
+        timeout: float | None = None,
+    ) -> bytes:
+        """POST one raw binary chunk to the authenticated `/file.cgi` family."""
+
+        if not isinstance(data, bytes):
+            raise TypeError("file upload data must be bytes")
+        effective_timeout = self.timeout if timeout is None else timeout
+        response = self._request(
+            "POST",
+            self.file_url,
+            params=dict(params),
+            data=data,
+            headers={"Content-Type": "application/octet-stream"},
+            timeout=effective_timeout,
+        )
+        return bytes(response.content)
+
     def close(self) -> None:
         self.session.close()
 

tests/conftest.py

@@ -6,10 +6,22 @@
 
 
 class FakeResponse:
-    def __init__(self, payload: Any, *, status_code: int = 200, on_json=None) -> None:
+    def __init__(
+        self,
+        payload: Any = None,
+        *,
+        status_code: int = 200,
+        on_json=None,
+        content: bytes | None = None,
+        headers: dict[str, str] | None = None,
+    ) -> None:
         self._payload = payload
         self.status_code = status_code
         self._on_json = on_json
+        if content is None:
+            content = payload if isinstance(payload, bytes) else b""
+        self.content = bytes(content)
+        self.headers = dict(headers or {})
 
     def raise_for_status(self) -> None:
         if self.status_code >= 400:
@@ -24,7 +36,7 @@ def json(self) -> Any:
 
 
 class FakeSession:
-    def __init__(self, responses: list[FakeResponse]) -> None:
+    def __init__(self, responses: list[Any]) -> None:
         self.responses = list(responses)
         self.calls: list[tuple[str, str, dict[str, Any]]] = []
         self.cookies = requests.cookies.RequestsCookieJar()
@@ -34,7 +46,10 @@ def request(self, method: str, url: str, **kwargs: Any) -> FakeResponse:
         self.calls.append((method, url, kwargs))
         if not self.responses:
             raise AssertionError("unexpected request")
-        return self.responses.pop(0)
+        response = self.responses.pop(0)
+        if isinstance(response, Exception):
+            raise response
+        return response
 
     def close(self) -> None:
         self.closed = True