fix(signing): Improve filenames and add tests.

Author: amilcarlucas

ardupilot_methodic_configurator/backend_flightcontroller.py

@@ -479,7 +479,7 @@ def is_battery_monitoring_enabled(self) -> bool:
 
     # MAVLink Signing Functionality
 
-    def setup_signing(
+    def setup_signing(  # pylint: disable=too-many-arguments, too-many-positional-arguments
         self,
         key: bytes,
         sign_outgoing: bool = True,

ardupilot_methodic_configurator/backend_signing_keystore.py

@@ -27,6 +27,12 @@
 KEYSTORE_FILENAME = "mavlink_signing_keys.json"
 KEYSTORE_VERSION = 1
 
+# PBKDF2 iteration counts
+# File storage uses lower iterations as vehicle_id provides limited entropy
+PBKDF2_ITERATIONS_FILE = 480_000
+# Export/import uses higher iterations for password-based encryption
+PBKDF2_ITERATIONS_EXPORT = 600_000
+
 
 @dataclass
 class StoredKey:
@@ -39,12 +45,12 @@ class StoredKey:
     salt: str  # Base64 encoded salt for encryption
     description: str = ""
 
-    def to_dict(self) -> dict:
+    def to_dict(self) -> dict[str, str]:
         """Convert to dictionary for JSON serialization."""
         return asdict(self)
 
     @classmethod
-    def from_dict(cls, data: dict) -> "StoredKey":
+    def from_dict(cls, data: dict[str, str]) -> "StoredKey":
         """Create from dictionary."""
         return cls(**data)
 
@@ -56,18 +62,19 @@ class KeystoreData:
     version: int = KEYSTORE_VERSION
     keys: dict[str, StoredKey] = field(default_factory=dict)
 
-    def to_dict(self) -> dict:
+    def to_dict(self) -> dict[str, object]:
         """Convert to dictionary for JSON serialization."""
         return {
             "version": self.version,
             "keys": {k: v.to_dict() for k, v in self.keys.items()},
         }
 
     @classmethod
-    def from_dict(cls, data: dict) -> "KeystoreData":
+    def from_dict(cls, data: dict[str, object]) -> "KeystoreData":
         """Create from dictionary."""
-        keys = {k: StoredKey.from_dict(v) for k, v in data.get("keys", {}).items()}
-        return cls(version=data.get("version", KEYSTORE_VERSION), keys=keys)
+        keys_data: dict[str, dict[str, str]] = data.get("keys", {})  # type: ignore[assignment]
+        keys = {k: StoredKey.from_dict(v) for k, v in keys_data.items()}
+        return cls(version=data.get("version", KEYSTORE_VERSION), keys=keys)  # type: ignore[arg-type]
 
 
 class SigningKeystore:
@@ -84,6 +91,14 @@ class SigningKeystore:
     - Password-protected export/import
     - Encrypted storage with key derivation
 
+    **IMPORTANT SECURITY LIMITATION:**
+    The file-based fallback storage uses vehicle_id as the encryption key derivation input.
+    This provides obfuscation but NOT strong encryption against attackers with filesystem access.
+    For production use, consider:
+    - Using the OS keyring when available (provides proper security)
+    - Implementing additional master password protection
+    - Restricting filesystem access to the keystore file
+
     Attributes:
         keyring_available: Whether OS keyring is available
         fallback_path: Path to encrypted fallback file
@@ -141,7 +156,10 @@ def _check_keyring_available(self) -> bool:
             # Test with a dummy operation
             test_key = f"_test_keyring_{secrets.token_hex(4)}"
             keyring.set_password(self._keyring_service, test_key, "test")
-            keyring.delete_password(self._keyring_service, test_key)
+            try:
+                keyring.delete_password(self._keyring_service, test_key)
+            except Exception as cleanup_exc:  # pylint: disable=broad-except
+                self._logger.warning("Failed to cleanup keyring test: %s", cleanup_exc)
             return True
         except ImportError:
             self._logger.debug("Keyring package not installed")
@@ -202,13 +220,13 @@ def store_key(self, vehicle_id: str, key: bytes, description: str = "") -> bool:
         # Fall back to encrypted file storage
         return self._store_key_in_file(vehicle_id, key, description)
 
-    def _store_key_in_file(self, vehicle_id: str, key: bytes, description: str = "") -> bool:
+    def _store_key_in_file(self, vehicle_id: str, key: bytes, description: str = "") -> bool:  # pylint: disable=too-many-locals
         """Store key in encrypted file."""
         try:
             from cryptography.fernet import Fernet  # noqa: PLC0415 # pylint: disable=import-outside-toplevel
             from cryptography.hazmat.primitives import hashes  # noqa: PLC0415 # pylint: disable=import-outside-toplevel
-            from cryptography.hazmat.primitives.kdf.pbkdf2 import (  # noqa: PLC0415
-                PBKDF2HMAC,  # pylint: disable=import-outside-toplevel
+            from cryptography.hazmat.primitives.kdf.pbkdf2 import (  # noqa: PLC0415 # pylint: disable=import-outside-toplevel
+                PBKDF2HMAC,
             )
 
             # Load existing keystore or create new
@@ -223,7 +241,7 @@ def _store_key_in_file(self, vehicle_id: str, key: bytes, description: str = "")
                 algorithm=hashes.SHA256(),
                 length=32,
                 salt=salt,
-                iterations=480000,
+                iterations=PBKDF2_ITERATIONS_FILE,
             )
             derived_key = kdf.derive(vehicle_id.encode())
             fernet_key = base64.urlsafe_b64encode(derived_key)
@@ -286,8 +304,8 @@ def _retrieve_key_from_file(self, vehicle_id: str) -> Optional[bytes]:
         try:
             from cryptography.fernet import Fernet  # noqa: PLC0415 # pylint: disable=import-outside-toplevel
             from cryptography.hazmat.primitives import hashes  # noqa: PLC0415 # pylint: disable=import-outside-toplevel
-            from cryptography.hazmat.primitives.kdf.pbkdf2 import (  # noqa: PLC0415
-                PBKDF2HMAC,  # pylint: disable=import-outside-toplevel
+            from cryptography.hazmat.primitives.kdf.pbkdf2 import (  # noqa: PLC0415 # pylint: disable=import-outside-toplevel
+                PBKDF2HMAC,
             )
 
             keystore = self._load_keystore_file()
@@ -306,14 +324,15 @@ def _retrieve_key_from_file(self, vehicle_id: str) -> Optional[bytes]:
                 algorithm=hashes.SHA256(),
                 length=32,
                 salt=salt,
-                iterations=480000,
+                iterations=PBKDF2_ITERATIONS_FILE,
             )
             derived_key = kdf.derive(vehicle_id.encode())
             fernet_key = base64.urlsafe_b64encode(derived_key)
             fernet = Fernet(fernet_key)
 
             # Decrypt the signing key
-            return fernet.decrypt(encrypted_key)
+            decrypted_key: bytes = fernet.decrypt(encrypted_key)
+            return decrypted_key
 
         except Exception as exc:  # pylint: disable=broad-except
             self._logger.exception("Failed to retrieve key from file: %s", exc)
@@ -397,8 +416,8 @@ def export_key(self, vehicle_id: str, password: str) -> Optional[str]:
         try:
             from cryptography.fernet import Fernet  # noqa: PLC0415 # pylint: disable=import-outside-toplevel
             from cryptography.hazmat.primitives import hashes  # noqa: PLC0415 # pylint: disable=import-outside-toplevel
-            from cryptography.hazmat.primitives.kdf.pbkdf2 import (  # noqa: PLC0415
-                PBKDF2HMAC,  # pylint: disable=import-outside-toplevel
+            from cryptography.hazmat.primitives.kdf.pbkdf2 import (  # noqa: PLC0415  # pylint: disable=import-outside-toplevel
+                PBKDF2HMAC,
             )
 
             # Generate salt
@@ -409,7 +428,7 @@ def export_key(self, vehicle_id: str, password: str) -> Optional[str]:
                 algorithm=hashes.SHA256(),
                 length=32,
                 salt=salt,
-                iterations=600000,  # Higher iterations for password-based
+                iterations=PBKDF2_ITERATIONS_EXPORT,
             )
             derived_key = kdf.derive(password.encode())
             fernet_key = base64.urlsafe_b64encode(derived_key)
@@ -433,7 +452,7 @@ def export_key(self, vehicle_id: str, password: str) -> Optional[str]:
             self._logger.exception("Failed to export key: %s", exc)
             return None
 
-    def import_key(self, export_data: str, password: str) -> Optional[str]:
+    def import_key(self, export_data: str, password: str) -> Optional[str]:  # pylint: disable=too-many-locals
         """
         Import a signing key from an encrypted export.
 
@@ -448,8 +467,8 @@ def import_key(self, export_data: str, password: str) -> Optional[str]:
         try:
             from cryptography.fernet import Fernet, InvalidToken  # noqa: PLC0415 # pylint: disable=import-outside-toplevel
             from cryptography.hazmat.primitives import hashes  # noqa: PLC0415 # pylint: disable=import-outside-toplevel
-            from cryptography.hazmat.primitives.kdf.pbkdf2 import (  # noqa: PLC0415
-                PBKDF2HMAC,  # pylint: disable=import-outside-toplevel
+            from cryptography.hazmat.primitives.kdf.pbkdf2 import (  # noqa: PLC0415  # pylint: disable=import-outside-toplevel
+                PBKDF2HMAC,
             )
 
             # Decode export package
@@ -464,7 +483,7 @@ def import_key(self, export_data: str, password: str) -> Optional[str]:
                 algorithm=hashes.SHA256(),
                 length=32,
                 salt=salt,
-                iterations=600000,
+                iterations=PBKDF2_ITERATIONS_EXPORT,
             )
             derived_key = kdf.derive(password.encode())
             fernet_key = base64.urlsafe_b64encode(derived_key)
@@ -495,19 +514,24 @@ def _load_keystore_file(self) -> KeystoreData:
             with open(self._fallback_path, encoding="utf-8") as f:
                 data = json.load(f)
             return KeystoreData.from_dict(data)
-        except Exception as exc:  # pylint: disable=broad-except
+        except (json.JSONDecodeError, OSError, KeyError, TypeError, ValueError) as exc:
             self._logger.warning("Failed to load keystore file: %s", exc)
             return KeystoreData()
 
     def _save_keystore_file(self, keystore: KeystoreData) -> None:
-        """Save keystore to file."""
+        """Save keystore to file with file locking for concurrent access protection."""
         # Ensure directory exists
         self._fallback_path.parent.mkdir(parents=True, exist_ok=True)
 
         # Write atomically using a temporary file
         temp_path = self._fallback_path.with_suffix(".tmp")
         try:
             with open(temp_path, "w", encoding="utf-8") as f:
+                # Apply file lock on Unix systems to prevent concurrent writes
+                if os.name != "nt":
+                    import fcntl  # noqa: PLC0415 # pylint: disable=import-outside-toplevel
+
+                    fcntl.flock(f.fileno(), fcntl.LOCK_EX)
                 json.dump(keystore.to_dict(), f, indent=2)
 
             # Atomic rename