""" Satellite Registration and Blacklist Management This module implements satellite registration file management and blacklist functionality for the Trixy satellite management system. It provides MAC-based registration file storage and comprehensive blacklist management as specified in CLAUDE.md. Key Features: - MAC-based registration file management (JSON format) - Satellite registration validation and persistence - Blacklist management with reason tracking - Thread-safe operations for concurrent access - Integration with satellite manager - Comprehensive logging and error handling - Automatic file backup and recovery """ import json import threading import shutil from typing import Dict, List, Optional, Any, Set, Tuple from dataclasses import dataclass, field from datetime import datetime, timezone from pathlib import Path from enum import Enum import os import uuid from .satellite import SatelliteInfo, SatelliteCapability def pprint(message: str) -> None: """Registration system logging function.""" print(f"[REGISTRATION] {message}") class RegistrationStatus(Enum): """Status of satellite registration.""" PENDING = "pending" APPROVED = "approved" REJECTED = "rejected" EXPIRED = "expired" REVOKED = "revoked" @dataclass class SatelliteRegistration: """ Satellite registration record with complete information. This class stores all information about a satellite registration, including approval status, timestamps, and metadata. """ satellite_info: SatelliteInfo status: RegistrationStatus = RegistrationStatus.PENDING registered_by: Optional[str] = None approved_by: Optional[str] = None approved_at: Optional[datetime] = None created_at: Optional[datetime] = field(default_factory=lambda: datetime.now(timezone.utc)) updated_at: Optional[datetime] = field(default_factory=lambda: datetime.now(timezone.utc)) expires_at: Optional[datetime] = None notes: str = "" metadata: Dict[str, Any] = field(default_factory=dict) def to_dict(self) -> Dict[str, Any]: """Convert registration to dictionary representation.""" return { "satellite_info": self.satellite_info.to_dict(), "status": self.status.value, "registered_by": self.registered_by, "approved_by": self.approved_by, "approved_at": self.approved_at.isoformat() if self.approved_at else None, "created_at": self.created_at.isoformat() if self.created_at else None, "updated_at": self.updated_at.isoformat() if self.updated_at else None, "expires_at": self.expires_at.isoformat() if self.expires_at else None, "notes": self.notes, "metadata": self.metadata, } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'SatelliteRegistration': """Create registration from dictionary representation.""" return cls( satellite_info=SatelliteInfo.from_dict(data["satellite_info"]), status=RegistrationStatus(data.get("status", "pending")), registered_by=data.get("registered_by"), approved_by=data.get("approved_by"), approved_at=datetime.fromisoformat(data["approved_at"]) if data.get("approved_at") else None, created_at=datetime.fromisoformat(data["created_at"]) if data.get("created_at") else None, updated_at=datetime.fromisoformat(data["updated_at"]) if data.get("updated_at") else None, expires_at=datetime.fromisoformat(data["expires_at"]) if data.get("expires_at") else None, notes=data.get("notes", ""), metadata=data.get("metadata", {}), ) def update_timestamp(self) -> None: """Update the updated_at timestamp.""" self.updated_at = datetime.now(timezone.utc) def approve(self, approved_by: Optional[str] = None, notes: str = "") -> None: """Approve this registration.""" self.status = RegistrationStatus.APPROVED self.approved_by = approved_by self.approved_at = datetime.now(timezone.utc) if notes: self.notes = notes self.update_timestamp() def reject(self, reason: str = "") -> None: """Reject this registration.""" self.status = RegistrationStatus.REJECTED if reason: self.notes = reason self.update_timestamp() def revoke(self, reason: str = "") -> None: """Revoke this registration.""" self.status = RegistrationStatus.REVOKED if reason: self.notes = reason self.update_timestamp() def is_active(self) -> bool: """Check if registration is currently active.""" if self.status != RegistrationStatus.APPROVED: return False if self.expires_at and datetime.now(timezone.utc) > self.expires_at: return False return True def is_expired(self) -> bool: """Check if registration has expired.""" return (self.expires_at is not None and datetime.now(timezone.utc) > self.expires_at) @dataclass class BlacklistEntry: """ Blacklist entry with reason and metadata. This class stores information about blacklisted MAC addresses, including the reason for blacklisting and tracking information. """ mac_address: str reason: str = "manual_blacklist" blacklisted_by: Optional[str] = None blacklisted_at: Optional[datetime] = field(default_factory=lambda: datetime.now(timezone.utc)) expires_at: Optional[datetime] = None notes: str = "" metadata: Dict[str, Any] = field(default_factory=dict) def to_dict(self) -> Dict[str, Any]: """Convert blacklist entry to dictionary representation.""" return { "mac_address": self.mac_address, "reason": self.reason, "blacklisted_by": self.blacklisted_by, "blacklisted_at": self.blacklisted_at.isoformat() if self.blacklisted_at else None, "expires_at": self.expires_at.isoformat() if self.expires_at else None, "notes": self.notes, "metadata": self.metadata, } @classmethod def from_dict(cls, data: Dict[str, Any]) -> 'BlacklistEntry': """Create blacklist entry from dictionary representation.""" return cls( mac_address=data["mac_address"], reason=data.get("reason", "manual_blacklist"), blacklisted_by=data.get("blacklisted_by"), blacklisted_at=datetime.fromisoformat(data["blacklisted_at"]) if data.get("blacklisted_at") else None, expires_at=datetime.fromisoformat(data["expires_at"]) if data.get("expires_at") else None, notes=data.get("notes", ""), metadata=data.get("metadata", {}), ) def is_active(self) -> bool: """Check if blacklist entry is currently active.""" if self.expires_at and datetime.now(timezone.utc) > self.expires_at: return False return True def is_expired(self) -> bool: """Check if blacklist entry has expired.""" return (self.expires_at is not None and datetime.now(timezone.utc) > self.expires_at) class RegistrationError(Exception): """Base exception for registration system errors.""" pass class RegistrationValidationError(RegistrationError): """Raised when registration validation fails.""" pass class RegistrationFileError(RegistrationError): """Raised when registration file operations fail.""" pass class BlacklistError(RegistrationError): """Raised when blacklist operations fail.""" pass class RegistrationManager: """ Manages satellite registration files and validation. This class handles the creation, validation, and persistence of satellite registration records. It maintains thread-safe access to registration files and provides comprehensive validation. """ def __init__( self, registration_dir: Optional[str] = None, backup_dir: Optional[str] = None, auto_backup: bool = True, max_backups: int = 10 ): """ Initialize the registration manager. Args: registration_dir: Directory for registration files backup_dir: Directory for backup files auto_backup: Enable automatic backups max_backups: Maximum number of backup files to keep """ self._registration_dir = Path(registration_dir or "config/satellites/registrations") self._backup_dir = Path(backup_dir or "config/satellites/backups") self._auto_backup = auto_backup self._max_backups = max_backups # Ensure directories exist self._registration_dir.mkdir(parents=True, exist_ok=True) if self._auto_backup: self._backup_dir.mkdir(parents=True, exist_ok=True) # Thread safety self._lock = threading.RLock() # Cache for loaded registrations self._registrations_cache: Dict[str, SatelliteRegistration] = {} self._cache_dirty = True pprint(f"RegistrationManager initialized (dir: {self._registration_dir})") def create_registration( self, satellite_info: SatelliteInfo, registered_by: Optional[str] = None, notes: str = "", auto_approve: bool = False, expires_in_days: Optional[int] = None ) -> SatelliteRegistration: """ Create a new satellite registration. Args: satellite_info: Complete satellite information registered_by: User who created the registration notes: Additional notes auto_approve: Automatically approve registration expires_in_days: Days until registration expires Returns: SatelliteRegistration: The created registration Raises: RegistrationValidationError: If validation fails """ with self._lock: # Validate satellite info self._validate_satellite_info(satellite_info) # Check if already registered existing = self.get_registration(satellite_info.mac_address) if existing and existing.is_active(): raise RegistrationValidationError( f"Satellite {satellite_info.mac_address} is already registered" ) # Calculate expiration expires_at = None if expires_in_days: from datetime import timedelta expires_at = datetime.now(timezone.utc) + timedelta(days=expires_in_days) # Create registration registration = SatelliteRegistration( satellite_info=satellite_info, status=RegistrationStatus.APPROVED if auto_approve else RegistrationStatus.PENDING, registered_by=registered_by, approved_by=registered_by if auto_approve else None, approved_at=datetime.now(timezone.utc) if auto_approve else None, expires_at=expires_at, notes=notes ) # Save to file self._save_registration(registration) # Update cache self._registrations_cache[satellite_info.mac_address.lower()] = registration status_text = "approved" if auto_approve else "pending" pprint(f"Registration created: {satellite_info.alias_name} ({satellite_info.mac_address}) - {status_text}") return registration def get_registration(self, mac_address: str) -> Optional[SatelliteRegistration]: """ Get registration for a MAC address. Args: mac_address: MAC address to look up Returns: Optional[SatelliteRegistration]: Registration if found """ with self._lock: mac_address = mac_address.lower() # Check cache first if not self._cache_dirty and mac_address in self._registrations_cache: return self._registrations_cache[mac_address] # Load from file registration = self._load_registration(mac_address) # Update cache if registration: self._registrations_cache[mac_address] = registration return registration def update_registration( self, mac_address: str, updates: Dict[str, Any], updated_by: Optional[str] = None ) -> bool: """ Update an existing registration. Args: mac_address: MAC address of registration to update updates: Dictionary of fields to update updated_by: User making the update Returns: bool: True if updated successfully """ with self._lock: registration = self.get_registration(mac_address) if not registration: return False # Backup before updating if self._auto_backup: self._create_backup(registration) # Apply updates for key, value in updates.items(): if key == "status" and isinstance(value, str): registration.status = RegistrationStatus(value) elif key == "satellite_info" and isinstance(value, dict): registration.satellite_info = SatelliteInfo.from_dict(value) elif hasattr(registration, key): setattr(registration, key, value) # Update metadata registration.metadata["updated_by"] = updated_by registration.update_timestamp() # Save to file self._save_registration(registration) # Update cache self._registrations_cache[mac_address.lower()] = registration pprint(f"Registration updated: {mac_address}") return True def approve_registration( self, mac_address: str, approved_by: Optional[str] = None, notes: str = "" ) -> bool: """ Approve a pending registration. Args: mac_address: MAC address to approve approved_by: User approving the registration notes: Approval notes Returns: bool: True if approved successfully """ with self._lock: registration = self.get_registration(mac_address) if not registration: return False if registration.status != RegistrationStatus.PENDING: pprint(f"Cannot approve registration {mac_address}: status is {registration.status.value}") return False # Backup before approving if self._auto_backup: self._create_backup(registration) # Approve registration.approve(approved_by, notes) # Save to file self._save_registration(registration) # Update cache self._registrations_cache[mac_address.lower()] = registration pprint(f"Registration approved: {mac_address} by {approved_by}") return True def reject_registration( self, mac_address: str, reason: str = "manual_rejection" ) -> bool: """ Reject a registration. Args: mac_address: MAC address to reject reason: Reason for rejection Returns: bool: True if rejected successfully """ with self._lock: registration = self.get_registration(mac_address) if not registration: return False # Backup before rejecting if self._auto_backup: self._create_backup(registration) # Reject registration.reject(reason) # Save to file self._save_registration(registration) # Update cache self._registrations_cache[mac_address.lower()] = registration pprint(f"Registration rejected: {mac_address} - {reason}") return True def revoke_registration( self, mac_address: str, reason: str = "manual_revocation" ) -> bool: """ Revoke an active registration. Args: mac_address: MAC address to revoke reason: Reason for revocation Returns: bool: True if revoked successfully """ with self._lock: registration = self.get_registration(mac_address) if not registration: return False # Backup before revoking if self._auto_backup: self._create_backup(registration) # Revoke registration.revoke(reason) # Save to file self._save_registration(registration) # Update cache self._registrations_cache[mac_address.lower()] = registration pprint(f"Registration revoked: {mac_address} - {reason}") return True def delete_registration(self, mac_address: str) -> bool: """ Permanently delete a registration. Args: mac_address: MAC address to delete Returns: bool: True if deleted successfully """ with self._lock: mac_address = mac_address.lower() registration = self.get_registration(mac_address) if registration and self._auto_backup: self._create_backup(registration) # Remove file filename = f"{mac_address.replace(':', '_')}.json" file_path = self._registration_dir / filename try: if file_path.exists(): file_path.unlink() pprint(f"Registration deleted: {mac_address}") # Remove from cache if mac_address in self._registrations_cache: del self._registrations_cache[mac_address] return True except Exception as e: pprint(f"Error deleting registration {mac_address}: {e}") return False def get_all_registrations(self) -> List[SatelliteRegistration]: """Get all registrations.""" with self._lock: self._refresh_cache() return list(self._registrations_cache.values()) def get_registrations_by_status(self, status: RegistrationStatus) -> List[SatelliteRegistration]: """Get registrations by status.""" with self._lock: registrations = self.get_all_registrations() return [reg for reg in registrations if reg.status == status] def get_active_registrations(self) -> List[SatelliteRegistration]: """Get all active (approved and not expired) registrations.""" with self._lock: registrations = self.get_all_registrations() return [reg for reg in registrations if reg.is_active()] def get_expired_registrations(self) -> List[SatelliteRegistration]: """Get all expired registrations.""" with self._lock: registrations = self.get_all_registrations() return [reg for reg in registrations if reg.is_expired()] def cleanup_expired_registrations(self) -> int: """ Clean up expired registrations. Returns: int: Number of registrations cleaned up """ with self._lock: expired = self.get_expired_registrations() cleaned = 0 for registration in expired: if registration.status == RegistrationStatus.APPROVED: registration.status = RegistrationStatus.EXPIRED registration.update_timestamp() self._save_registration(registration) cleaned += 1 if cleaned > 0: pprint(f"Cleaned up {cleaned} expired registrations") return cleaned def _validate_satellite_info(self, satellite_info: SatelliteInfo) -> None: """Validate satellite information.""" if not satellite_info.satellite_id: raise RegistrationValidationError("Satellite ID is required") if not satellite_info.mac_address: raise RegistrationValidationError("MAC address is required") if not satellite_info.room_id: raise RegistrationValidationError("Room ID is required") if not satellite_info.alias_name: raise RegistrationValidationError("Alias name is required") # Validate MAC address format from . import validate_mac_address if not validate_mac_address(satellite_info.mac_address): raise RegistrationValidationError(f"Invalid MAC address format: {satellite_info.mac_address}") def _load_registration(self, mac_address: str) -> Optional[SatelliteRegistration]: """Load registration from file.""" filename = f"{mac_address.replace(':', '_')}.json" file_path = self._registration_dir / filename if not file_path.exists(): return None try: with open(file_path, 'r') as f: data = json.load(f) return SatelliteRegistration.from_dict(data) except Exception as e: pprint(f"Error loading registration {mac_address}: {e}") return None def _save_registration(self, registration: SatelliteRegistration) -> None: """Save registration to file.""" mac_address = registration.satellite_info.mac_address.lower() filename = f"{mac_address.replace(':', '_')}.json" file_path = self._registration_dir / filename try: with open(file_path, 'w') as f: json.dump(registration.to_dict(), f, indent=2) except Exception as e: raise RegistrationFileError(f"Error saving registration {mac_address}: {e}") def _refresh_cache(self) -> None: """Refresh the registrations cache.""" if not self._cache_dirty: return self._registrations_cache.clear() for file_path in self._registration_dir.glob("*.json"): try: with open(file_path, 'r') as f: data = json.load(f) registration = SatelliteRegistration.from_dict(data) mac_address = registration.satellite_info.mac_address.lower() self._registrations_cache[mac_address] = registration except Exception as e: pprint(f"Error loading registration from {file_path}: {e}") self._cache_dirty = False def _create_backup(self, registration: SatelliteRegistration) -> None: """Create a backup of a registration.""" if not self._auto_backup: return mac_address = registration.satellite_info.mac_address.lower() timestamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") filename = f"{mac_address.replace(':', '_')}_{timestamp}.json" backup_path = self._backup_dir / filename try: with open(backup_path, 'w') as f: json.dump(registration.to_dict(), f, indent=2) # Clean up old backups self._cleanup_old_backups(mac_address) except Exception as e: pprint(f"Error creating backup for {mac_address}: {e}") def _cleanup_old_backups(self, mac_address: str) -> None: """Clean up old backup files.""" pattern = f"{mac_address.replace(':', '_')}_*.json" backup_files = list(self._backup_dir.glob(pattern)) if len(backup_files) > self._max_backups: # Sort by modification time and remove oldest backup_files.sort(key=lambda f: f.stat().st_mtime) for old_backup in backup_files[:-self._max_backups]: try: old_backup.unlink() except Exception as e: pprint(f"Error removing old backup {old_backup}: {e}") class BlacklistManager: """ Manages MAC address blacklist with comprehensive tracking. This class handles blacklisting of MAC addresses with detailed tracking of reasons, expiration, and metadata. """ def __init__( self, blacklist_file: Optional[str] = None, backup_file: Optional[str] = None, auto_backup: bool = True ): """ Initialize the blacklist manager. Args: blacklist_file: Path to blacklist file backup_file: Path to backup file auto_backup: Enable automatic backups """ self._blacklist_file = Path(blacklist_file or "config/satellites/blacklist.json") self._backup_file = Path(backup_file or "config/satellites/blacklist_backup.json") self._auto_backup = auto_backup # Ensure directory exists self._blacklist_file.parent.mkdir(parents=True, exist_ok=True) # Thread safety self._lock = threading.RLock() # Storage self._blacklist: Dict[str, BlacklistEntry] = {} # Load existing blacklist self._load_blacklist() pprint(f"BlacklistManager initialized (file: {self._blacklist_file})") def add_to_blacklist( self, mac_address: str, reason: str = "manual_blacklist", blacklisted_by: Optional[str] = None, notes: str = "", expires_in_days: Optional[int] = None ) -> bool: """ Add a MAC address to the blacklist. Args: mac_address: MAC address to blacklist reason: Reason for blacklisting blacklisted_by: User who added to blacklist notes: Additional notes expires_in_days: Days until blacklist expires Returns: bool: True if added successfully """ with self._lock: mac_address = mac_address.lower() # Backup before modifying if self._auto_backup: self._create_backup() # Calculate expiration expires_at = None if expires_in_days: from datetime import timedelta expires_at = datetime.now(timezone.utc) + timedelta(days=expires_in_days) # Create blacklist entry entry = BlacklistEntry( mac_address=mac_address, reason=reason, blacklisted_by=blacklisted_by, expires_at=expires_at, notes=notes ) self._blacklist[mac_address] = entry # Save to file self._save_blacklist() pprint(f"MAC address blacklisted: {mac_address} - {reason}") return True def remove_from_blacklist(self, mac_address: str) -> bool: """ Remove a MAC address from the blacklist. Args: mac_address: MAC address to remove Returns: bool: True if removed successfully """ with self._lock: mac_address = mac_address.lower() if mac_address not in self._blacklist: return True # Backup before modifying if self._auto_backup: self._create_backup() del self._blacklist[mac_address] # Save to file self._save_blacklist() pprint(f"MAC address removed from blacklist: {mac_address}") return True def is_blacklisted(self, mac_address: str) -> bool: """Check if a MAC address is blacklisted.""" with self._lock: mac_address = mac_address.lower() entry = self._blacklist.get(mac_address) if not entry: return False # Check if expired if entry.is_expired(): # Automatically remove expired entries self.remove_from_blacklist(mac_address) return False return True def get_blacklist_entry(self, mac_address: str) -> Optional[BlacklistEntry]: """Get blacklist entry for a MAC address.""" with self._lock: return self._blacklist.get(mac_address.lower()) def get_all_blacklist_entries(self) -> List[BlacklistEntry]: """Get all blacklist entries.""" with self._lock: return list(self._blacklist.values()) def get_active_blacklist_entries(self) -> List[BlacklistEntry]: """Get all active (non-expired) blacklist entries.""" with self._lock: return [entry for entry in self._blacklist.values() if entry.is_active()] def get_expired_blacklist_entries(self) -> List[BlacklistEntry]: """Get all expired blacklist entries.""" with self._lock: return [entry for entry in self._blacklist.values() if entry.is_expired()] def cleanup_expired_entries(self) -> int: """ Clean up expired blacklist entries. Returns: int: Number of entries cleaned up """ with self._lock: expired = self.get_expired_blacklist_entries() for entry in expired: self.remove_from_blacklist(entry.mac_address) if expired: pprint(f"Cleaned up {len(expired)} expired blacklist entries") return len(expired) def get_blacklist_stats(self) -> Dict[str, Any]: """Get blacklist statistics.""" with self._lock: entries = list(self._blacklist.values()) active_entries = [e for e in entries if e.is_active()] expired_entries = [e for e in entries if e.is_expired()] # Count by reason reason_stats = {} for entry in entries: reason_stats[entry.reason] = reason_stats.get(entry.reason, 0) + 1 return { "total_entries": len(entries), "active_entries": len(active_entries), "expired_entries": len(expired_entries), "reason_breakdown": reason_stats, "file_path": str(self._blacklist_file), "last_updated": datetime.now(timezone.utc).isoformat(), } def _load_blacklist(self) -> None: """Load blacklist from file.""" if not self._blacklist_file.exists(): return try: with open(self._blacklist_file, 'r') as f: data = json.load(f) # Handle both old format (list) and new format (dict) if isinstance(data, list): # Old format: list of MAC addresses for mac_address in data: entry = BlacklistEntry( mac_address=mac_address.lower(), reason="legacy_blacklist" ) self._blacklist[mac_address.lower()] = entry elif isinstance(data, dict): if "entries" in data: # New format: entries list for entry_data in data["entries"]: entry = BlacklistEntry.from_dict(entry_data) self._blacklist[entry.mac_address.lower()] = entry elif "blacklisted_macs" in data: # Intermediate format: simple list for mac_address in data["blacklisted_macs"]: entry = BlacklistEntry( mac_address=mac_address.lower(), reason="legacy_blacklist" ) self._blacklist[mac_address.lower()] = entry if self._blacklist: pprint(f"Loaded {len(self._blacklist)} blacklist entries") except Exception as e: pprint(f"Error loading blacklist: {e}") def _save_blacklist(self) -> None: """Save blacklist to file.""" try: data = { "version": "1.0", "updated_at": datetime.now(timezone.utc).isoformat(), "entries": [entry.to_dict() for entry in self._blacklist.values()] } with open(self._blacklist_file, 'w') as f: json.dump(data, f, indent=2) except Exception as e: raise BlacklistError(f"Error saving blacklist: {e}") def _create_backup(self) -> None: """Create a backup of the blacklist.""" if not self._auto_backup or not self._blacklist_file.exists(): return try: shutil.copy2(self._blacklist_file, self._backup_file) except Exception as e: pprint(f"Error creating blacklist backup: {e}") # Factory functions def create_registration_manager(**kwargs) -> RegistrationManager: """Create a RegistrationManager instance.""" return RegistrationManager(**kwargs) def create_blacklist_manager(**kwargs) -> BlacklistManager: """Create a BlacklistManager instance.""" return BlacklistManager(**kwargs) # Module exports __all__ = [ "SatelliteRegistration", "RegistrationManager", "RegistrationError", "RegistrationValidationError", "RegistrationFileError", "RegistrationStatus", "BlacklistEntry", "BlacklistManager", "BlacklistError", "create_registration_manager", "create_blacklist_manager", ]