Fix all mypy type errors (10 errors in 5 files)
- Use type: ignore for cbor2/json Any returns in serialization/deadman - Fix callable→Callable in killswitch.py and usb_monitor.py - Add Ed25519PrivateKey assertion in CLI chain-wrap path - Allow None for RotationResult fingerprints - Annotate channel key as str in manager.py Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
This commit is contained in:
parent
17147856d1
commit
384d7e89d9
@ -282,6 +282,7 @@ def _attest_file(
|
|||||||
"""
|
"""
|
||||||
import hashlib
|
import hashlib
|
||||||
|
|
||||||
|
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
|
||||||
from cryptography.hazmat.primitives.serialization import (
|
from cryptography.hazmat.primitives.serialization import (
|
||||||
Encoding,
|
Encoding,
|
||||||
PublicFormat,
|
PublicFormat,
|
||||||
@ -333,6 +334,7 @@ def _attest_file(
|
|||||||
|
|
||||||
priv_pem = IDENTITY_PRIVATE_KEY.read_bytes()
|
priv_pem = IDENTITY_PRIVATE_KEY.read_bytes()
|
||||||
chain_key = load_pem_private_key(priv_pem, password=None)
|
chain_key = load_pem_private_key(priv_pem, password=None)
|
||||||
|
assert isinstance(chain_key, Ed25519PrivateKey)
|
||||||
|
|
||||||
chain_metadata: dict = {}
|
chain_metadata: dict = {}
|
||||||
if caption:
|
if caption:
|
||||||
|
|||||||
@ -51,7 +51,7 @@ def canonical_bytes(record: AttestationChainRecord) -> bytes:
|
|||||||
8: _entropy_to_cbor_map(record.entropy_witnesses) if record.entropy_witnesses else {},
|
8: _entropy_to_cbor_map(record.entropy_witnesses) if record.entropy_witnesses else {},
|
||||||
9: record.signer_pubkey,
|
9: record.signer_pubkey,
|
||||||
}
|
}
|
||||||
return cbor2.dumps(m, canonical=True)
|
return cbor2.dumps(m, canonical=True) # type: ignore[no-any-return]
|
||||||
|
|
||||||
|
|
||||||
def compute_record_hash(record: AttestationChainRecord) -> bytes:
|
def compute_record_hash(record: AttestationChainRecord) -> bytes:
|
||||||
@ -74,7 +74,7 @@ def serialize_record(record: AttestationChainRecord) -> bytes:
|
|||||||
9: record.signer_pubkey,
|
9: record.signer_pubkey,
|
||||||
10: record.signature,
|
10: record.signature,
|
||||||
}
|
}
|
||||||
return cbor2.dumps(m, canonical=True)
|
return cbor2.dumps(m, canonical=True) # type: ignore[no-any-return]
|
||||||
|
|
||||||
|
|
||||||
def deserialize_record(data: bytes) -> AttestationChainRecord:
|
def deserialize_record(data: bytes) -> AttestationChainRecord:
|
||||||
|
|||||||
@ -26,7 +26,7 @@ class DeadmanSwitch:
|
|||||||
def _load_state(self) -> dict:
|
def _load_state(self) -> dict:
|
||||||
if self._state_file.exists():
|
if self._state_file.exists():
|
||||||
with open(self._state_file) as f:
|
with open(self._state_file) as f:
|
||||||
return json.load(f)
|
return json.load(f) # type: ignore[no-any-return]
|
||||||
return {
|
return {
|
||||||
"armed": False,
|
"armed": False,
|
||||||
"last_checkin": None,
|
"last_checkin": None,
|
||||||
@ -64,7 +64,7 @@ class DeadmanSwitch:
|
|||||||
logger.info("Dead man's switch check-in recorded")
|
logger.info("Dead man's switch check-in recorded")
|
||||||
|
|
||||||
def is_armed(self) -> bool:
|
def is_armed(self) -> bool:
|
||||||
return self._load_state()["armed"]
|
return self._load_state()["armed"] # type: ignore[no-any-return]
|
||||||
|
|
||||||
def is_overdue(self) -> bool:
|
def is_overdue(self) -> bool:
|
||||||
"""Check if the switch has expired (past interval, ignoring grace)."""
|
"""Check if the switch has expired (past interval, ignoring grace)."""
|
||||||
|
|||||||
@ -15,6 +15,7 @@ import logging
|
|||||||
import platform
|
import platform
|
||||||
import shutil
|
import shutil
|
||||||
import subprocess
|
import subprocess
|
||||||
|
from collections.abc import Callable
|
||||||
from dataclasses import dataclass, field
|
from dataclasses import dataclass, field
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
@ -84,7 +85,7 @@ def execute_purge(scope: PurgeScope = PurgeScope.ALL, reason: str = "manual") ->
|
|||||||
result = PurgeResult()
|
result = PurgeResult()
|
||||||
logger.warning("KILLSWITCH ACTIVATED — reason: %s, scope: %s", reason, scope.value)
|
logger.warning("KILLSWITCH ACTIVATED — reason: %s, scope: %s", reason, scope.value)
|
||||||
|
|
||||||
steps: list[tuple[str, callable]] = [
|
steps: list[tuple[str, Callable]] = [
|
||||||
("destroy_identity_keys", lambda: _secure_delete_dir(paths.IDENTITY_DIR)),
|
("destroy_identity_keys", lambda: _secure_delete_dir(paths.IDENTITY_DIR)),
|
||||||
("destroy_channel_key", lambda: _secure_delete_file(paths.CHANNEL_KEY_FILE)),
|
("destroy_channel_key", lambda: _secure_delete_file(paths.CHANNEL_KEY_FILE)),
|
||||||
("destroy_flask_secret", lambda: _secure_delete_file(paths.INSTANCE_DIR / ".secret_key")),
|
("destroy_flask_secret", lambda: _secure_delete_file(paths.INSTANCE_DIR / ".secret_key")),
|
||||||
@ -142,7 +143,7 @@ except ImportError:
|
|||||||
def watch_hardware_button(
|
def watch_hardware_button(
|
||||||
pin: int = 17,
|
pin: int = 17,
|
||||||
hold_seconds: float = 5.0,
|
hold_seconds: float = 5.0,
|
||||||
callback: callable | None = None,
|
callback: Callable | None = None,
|
||||||
) -> None:
|
) -> None:
|
||||||
"""
|
"""
|
||||||
Monitor GPIO pin for physical killswitch button.
|
Monitor GPIO pin for physical killswitch button.
|
||||||
|
|||||||
@ -9,7 +9,9 @@ from __future__ import annotations
|
|||||||
|
|
||||||
import json
|
import json
|
||||||
import logging
|
import logging
|
||||||
|
from collections.abc import Callable
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
from typing import Any
|
||||||
|
|
||||||
from soosef.paths import USB_WHITELIST
|
from soosef.paths import USB_WHITELIST
|
||||||
|
|
||||||
@ -44,13 +46,13 @@ def save_whitelist(devices: set[str], path: Path | None = None) -> None:
|
|||||||
class USBMonitor:
|
class USBMonitor:
|
||||||
"""Watch for USB device connections and check against whitelist."""
|
"""Watch for USB device connections and check against whitelist."""
|
||||||
|
|
||||||
def __init__(self, on_violation: callable | None = None, whitelist_path: Path | None = None):
|
def __init__(self, on_violation: Callable | None = None, whitelist_path: Path | None = None):
|
||||||
if not HAS_PYUDEV:
|
if not HAS_PYUDEV:
|
||||||
raise RuntimeError("pyudev not available — USB monitoring requires Linux + pyudev")
|
raise RuntimeError("pyudev not available — USB monitoring requires Linux + pyudev")
|
||||||
|
|
||||||
self.whitelist = load_whitelist(whitelist_path)
|
self.whitelist = load_whitelist(whitelist_path)
|
||||||
self.on_violation = on_violation or self._default_violation
|
self.on_violation = on_violation or self._default_violation
|
||||||
self._observer = None
|
self._observer: Any = None
|
||||||
|
|
||||||
def start(self) -> None:
|
def start(self) -> None:
|
||||||
"""Start monitoring USB events in a background thread."""
|
"""Start monitoring USB events in a background thread."""
|
||||||
|
|||||||
@ -217,7 +217,7 @@ class KeystoreManager:
|
|||||||
"""Generate and store a new channel key."""
|
"""Generate and store a new channel key."""
|
||||||
from stegasoo import generate_channel_key
|
from stegasoo import generate_channel_key
|
||||||
|
|
||||||
key = generate_channel_key()
|
key: str = generate_channel_key()
|
||||||
self.set_channel_key(key)
|
self.set_channel_key(key)
|
||||||
return key
|
return key
|
||||||
|
|
||||||
|
|||||||
@ -29,6 +29,6 @@ class KeystoreStatus:
|
|||||||
class RotationResult:
|
class RotationResult:
|
||||||
"""Result of a key rotation operation."""
|
"""Result of a key rotation operation."""
|
||||||
|
|
||||||
old_fingerprint: str
|
old_fingerprint: str | None
|
||||||
new_fingerprint: str
|
new_fingerprint: str | None
|
||||||
archive_path: Path
|
archive_path: Path
|
||||||
|
|||||||
Loading…
Reference in New Issue
Block a user