Unleashed 200.15 support

aioruckus/backupsession.py

@@ -4,6 +4,10 @@
 import io
 import struct
 import tarfile
+
+from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
+from cryptography.hazmat.backends import default_backend
+from os import SEEK_CUR
 from typing import Any, Mapping, TYPE_CHECKING
 
 from .abcsession import AbcSession, ConfigItem
@@ -33,19 +37,64 @@ def __exit__(self, *exc: Any) -> None:
 
     def open_backup(self, backup_path: str) -> io.BytesIO:
         """Return the decrypted backup bytes"""
+        with open(backup_path, "rb") as backup_file:
+            magic = backup_file.read(4)
+            if magic == b'RKSF':
+                return self.open_commscope_backup(backup_file)
+            else:
+                backup_file.seek(0)
+                return self.open_tac_backup(backup_file)
+
+    @classmethod
+    def decrypt_key(cls, cipher_bytes: bytes) -> bytes:
+        padded_key = pow(int.from_bytes(cipher_bytes, 'big'), 65537, 23559046888044776627569879690471525499427612616504460325607886880157810091042540109382540840072568820382270758180649018860535002041926018790203547085546162549326945200443019963900872654422143820799219291504478283808912964667353808795633808052022964371726410677357834881346022671448243831605466569511830964339444687659616502868745663064525218488470606514409811838671765944249166136071060850237167429125523755638111097424494275181385870987411479009552515816450089719197508371290305110717762578033949377936003949760003095430389967102852124783026450284389704957901428442687247403657819155956894836033683283023293306459081).to_bytes(256, 'big')
+        return padded_key[padded_key.index(b'\x00', 2) + 1:]
+
+    def open_tac_backup(self, backup_file: io.BufferedReader) -> io.BytesIO:
+        """Return the decrypted TAC backup file"""
         (xor_int, xor_flip) = struct.unpack('QQ', b')\x1aB\x05\xbd,\xd6\xf25\xad\xb8\xe0?T\xc58')
         struct_int8 = struct.Struct('Q')
-        with open(backup_path, 'rb') as backup_file:
-            output_file = io.BytesIO()
-            input_data = backup_file.read()
-            previous_input_int = 0
-            for input_int in struct.unpack_from(str(len(input_data) // 8) + 'Q', input_data):
-                output_bytes = struct_int8.pack(previous_input_int ^ xor_int ^ input_int)
-                xor_int ^= xor_flip
-                previous_input_int = input_int
-                output_file.write(output_bytes)
-            output_file.seek(0)
-            return output_file
+        output_file = io.BytesIO()
+        previous_input_int = 0
+        input_data = backup_file.read()
+        for input_int in struct.unpack_from(str(len(input_data) // 8) + 'Q', input_data):
+            output_bytes = struct_int8.pack(previous_input_int ^ xor_int ^ input_int)
+            xor_int ^= xor_flip
+            previous_input_int = input_int
+            output_file.write(output_bytes)
+        output_file.seek(0)
+        return output_file
+
+    @classmethod
+    def skip_block(cls, backup_file: io.BufferedReader) -> None:
+        backup_file.seek(1, SEEK_CUR)
+        block_length = int.from_bytes(backup_file.read(4), byteorder='big', signed=False)
+        backup_file.seek(block_length, SEEK_CUR)
+
+    @classmethod
+    def get_block_length(cls, backup_file: io.BufferedReader) -> bytes:
+        backup_file.seek(1, SEEK_CUR)
+        return int.from_bytes(backup_file.read(4), byteorder='big', signed=False)
+
+    def open_commscope_backup(self, backup_file: io.BufferedReader) -> io.BytesIO:
+        """Return the decrypted CommScope Content Manager backup file"""
+        backup_file.seek(4, SEEK_CUR)
+        encrypted_key = backup_file.read(self.get_block_length(backup_file))
+        key = self.decrypt_key(encrypted_key)
+
+        self.skip_block(backup_file) # digest
+        self.skip_block(backup_file) # signature
+
+        decrypted_length = self.get_block_length(backup_file)
+        encrypted_bytes = backup_file.read()
+        cipher = Cipher(algorithms.AES(key), modes.ECB(), backend=default_backend())
+        decryptor = cipher.decryptor()
+        decrypted_bytes = decryptor.update(encrypted_bytes) + decryptor.finalize()
+
+        output_file = io.BytesIO()
+        output_file.write(decrypted_bytes[:decrypted_length])
+        output_file.seek(0)
+        return output_file
 
     @classmethod
     def create(cls, backup_path: str) -> "BackupSession":

requirements.txt

@@ -1,2 +1,3 @@
 aiohttp>=3.8.4
-xmltodict>=0.13.0
+xmltodict>=0.13.0
+cryptography>=41.0.0

setup.cfg

@@ -1,6 +1,6 @@
 [metadata]
 name = aioruckus
-version = 0.37
+version = 0.38
 author = ms264556
 author_email = [email protected]
 description = Python API to interact with Ruckus Unleashed and ZoneDirector devices.
@@ -19,6 +19,7 @@ include_package_data = True
 install_requires =
     aiohttp >= 3.8.4
     xmltodict >= 0.13.0
+    cryptography >= 41.0.0
 
 [options.packages.find]
 exclude =