# Anyone can send commands - no authentication
control.create_channel(ssrc=12345, frequency_hz=14.074e6)

import hmac
import hashlib

class AuthenticatedRadiodControl(RadiodControl):
    def __init__(self, status_address, shared_secret=None):
        super().__init__(status_address)
        self.shared_secret = shared_secret
    
    def send_command(self, cmdbuffer):
        if self.shared_secret:
            # Add HMAC signature
            signature = hmac.new(
                self.shared_secret.encode(),
                cmdbuffer,
                hashlib.sha256
            ).digest()
            cmdbuffer.extend(signature)
        super().send_command(cmdbuffer)

# control.py line 708, 814, 984
encode_int(cmdbuffer, StatusType.COMMAND_TAG, random.randint(1, 2**31))

import secrets

# Use cryptographically secure random
encode_int(cmdbuffer, StatusType.COMMAND_TAG, secrets.randbits(31))

def encode_string(buf: bytearray, type_val: int, s: str) -> int:
    s_bytes = s.encode('utf-8')
    # No validation on content, only length check
    if length >= 65536:
        raise ValueError(f"String too long: {length} bytes")

def validate_preset_name(preset: str) -> None:
    """Validate preset name is safe"""
    if not preset:
        raise ValidationError("Preset name cannot be empty")
    if len(preset) > 32:
        raise ValidationError(f"Preset name too long: {len(preset)} chars (max 32)")
    # Only allow alphanumeric, dash, underscore
    if not re.match(r'^[a-zA-Z0-9_-]+$', preset):
        raise ValidationError(f"Invalid preset name: {preset}")
    # Reject control characters
    if any(ord(c) < 32 for c in preset):
        raise ValidationError("Preset name contains control characters")

def encode_string(buf: bytearray, type_val: int, s: str, max_length: int = 65535) -> int:
    if len(s) > max_length:
        raise ValidationError(f"String too long: {len(s)} > {max_length}")
    # ... rest of encoding

import time
from collections import deque

class RateLimitedRadiodControl(RadiodControl):
    def __init__(self, status_address, max_commands_per_sec=100):
        super().__init__(status_address)
        self.max_commands_per_sec = max_commands_per_sec
        self.command_times = deque(maxlen=max_commands_per_sec)
        self._rate_limit_lock = threading.Lock()
    
    def send_command(self, cmdbuffer, **kwargs):
        with self._rate_limit_lock:
            now = time.time()
            # Remove timestamps older than 1 second
            while self.command_times and now - self.command_times[0] > 1.0:
                self.command_times.popleft()
            
            # Check if we've exceeded rate limit
            if len(self.command_times) >= self.max_commands_per_sec:
                sleep_time = 1.0 - (now - self.command_times[0])
                if sleep_time > 0:
                    logger.warning(f"Rate limit reached, sleeping {sleep_time:.3f}s")
                    time.sleep(sleep_time)
            
            self.command_times.append(now)
            return super().send_command(cmdbuffer, **kwargs)

# discovery.py line 107
status_sock = _create_status_listener_socket(multicast_addr)
# If error occurs before finally block, socket may leak

def discover_channels_native(status_address: str, listen_duration: float = 2.0):
    status_sock = None
    try:
        multicast_addr = resolve_multicast_address(status_address, timeout=2.0)
        status_sock = _create_status_listener_socket(multicast_addr)
        # ... operations ...
    except Exception as e:
        logger.error(f"Error: {e}")
        raise
    finally:
        if status_sock:
            try:
                status_sock.close()
            except Exception as e:
                logger.warning(f"Error closing socket: {e}")

def decode_double(data: bytes, length: int) -> float:
    # Reconstruct 8-byte big-endian representation
    value_bytes = b'\x00' * (8 - length) + data[:length]
    # What if length > 8? Buffer overflow in slice operation
    return struct.unpack('>d', value_bytes)[0]

def decode_double(data: bytes, length: int) -> float:
    """Decode a double with bounds checking"""
    if length > 8:
        logger.warning(f"Invalid double length: {length} (max 8)")
        length = 8  # Truncate to prevent issues
    if length < 0:
        raise ValidationError(f"Negative length: {length}")
    
    value_bytes = b'\x00' * (8 - length) + data[:length]
    return struct.unpack('>d', value_bytes)[0]

class RadiodControl:
    def __init__(self, status_address, 
                 connect_timeout=5.0,
                 command_timeout=2.0,
                 status_timeout=5.0):
        self.connect_timeout = connect_timeout
        self.command_timeout = command_timeout
        self.status_timeout = status_timeout

elif length == 10:
    # IPv6: 8 bytes address + 2 bytes port (big-endian)
    # Note: This is a truncated IPv6 address (only 8 bytes instead of 16)

class RadiodControl:
    def __init__(self, status_address):
        # ...
        self._last_status_time = None
        self._connection_state = "disconnected"
    
    def is_connected(self) -> bool:
        """Check if radiod is responding"""
        if self._last_status_time is None:
            return False
        return time.time() - self._last_status_time < 10.0
    
    def health_check(self) -> dict:
        """Perform health check"""
        try:
            channels = discover_channels(self.status_address, listen_duration=1.0)
            return {
                'connected': True,
                'channel_count': len(channels),
                'last_status': self._last_status_time
            }
        except Exception as e:
            return {'connected': False, 'error': str(e)}

import asyncio

class AsyncRadiodControl:
    async def create_channel_async(self, ssrc, frequency_hz, **kwargs):
        loop = asyncio.get_event_loop()
        return await loop.run_in_executor(
            None, 
            self.create_channel, 
            ssrc, 
            frequency_hz
        )
    
    async def tune_async(self, ssrc, **kwargs):
        # Proper async implementation with asyncio.DatagramProtocol
        pass

from dataclasses import dataclass, field
from typing import Dict
import time

@dataclass
class Metrics:
    commands_sent: int = 0
    commands_failed: int = 0
    status_received: int = 0
    avg_response_time: float = 0.0
    errors: Dict[str, int] = field(default_factory=dict)
    
    def record_command(self, success: bool, duration: float = 0.0):
        self.commands_sent += 1
        if not success:
            self.commands_failed += 1
        # Update avg response time with exponential moving average
        alpha = 0.1
        self.avg_response_time = (alpha * duration + 
                                  (1 - alpha) * self.avg_response_time)

class RadiodControl:
    def __init__(self, status_address):
        # ...
        self.metrics = Metrics()
    
    def get_metrics(self) -> dict:
        return {
            'commands_sent': self.metrics.commands_sent,
            'commands_failed': self.metrics.commands_failed,
            'success_rate': (self.metrics.commands_sent - self.metrics.commands_failed) / 
                           max(1, self.metrics.commands_sent),
            'avg_response_time_ms': self.metrics.avg_response_time * 1000,
        }

class ManagedChannel:
    """Managed channel with automatic cleanup"""
    def __init__(self, control, ssrc, **kwargs):
        self.control = control
        self.ssrc = ssrc
        self.created_at = time.time()
        self.last_used = time.time()
    
    def __enter__(self):
        self.control.create_channel(ssrc=self.ssrc, ...)
        return self
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        # Could send LOCK=0, frequency=0 to free channel
        pass

# Usage:
with ManagedChannel(control, ssrc=12345, frequency_hz=14.074e6) as ch:
    # Channel exists
    pass
# Channel automatically cleaned up

import configparser
from pathlib import Path

def load_config(config_file: Path = None) -> dict:
    """Load configuration from file"""
    config = configparser.ConfigParser()
    
    # Default locations
    config_paths = [
        Path.home() / '.ka9q' / 'config.ini',
        Path('/etc/ka9q/config.ini'),
        config_file,
    ]
    
    for path in config_paths:
        if path and path.exists():
            config.read(path)
            break
    
    return {
        'status_address': config.get('radiod', 'status_address', 
                                     fallback='radiod.local'),
        'timeout': config.getfloat('radiod', 'timeout', fallback=5.0),
        # ...
    }

# TODO: Extract decoder into standalone module to avoid this
temp_control = RadiodControl.__new__(RadiodControl)

# ka9q/protocol.py
def decode_status_packet(buffer: bytes) -> dict:
    """Standalone status packet decoder"""
    # Move _decode_status_response here
    pass

# discovery.py
from .protocol import decode_status_packet

def discover_channels_native(...):
    # ...
    status = decode_status_packet(buffer)

class BatchedRadiodControl(RadiodControl):
    def __init__(self, *args, batch_size=10, batch_timeout=0.1, **kwargs):
        super().__init__(*args, **kwargs)
        self.batch_size = batch_size
        self.batch_timeout = batch_timeout
        self._batch = []
        self._batch_lock = threading.Lock()
        self._batch_timer = None
    
    def _flush_batch(self):
        """Send all batched commands"""
        with self._batch_lock:
            if not self._batch:
                return
            
            # Combine into single multicast packet
            combined = bytearray()
            for cmd in self._batch:
                combined.extend(cmd)
            
            super().send_command(combined)
            self._batch.clear()
    
    def send_command(self, cmdbuffer, batch=True):
        if not batch:
            return super().send_command(cmdbuffer)
        
        with self._batch_lock:
            self._batch.append(cmdbuffer)
            
            # Flush if batch is full
            if len(self._batch) >= self.batch_size:
                self._flush_batch()
            else:
                # Set timer to flush after timeout
                if self._batch_timer:
                    self._batch_timer.cancel()
                self._batch_timer = threading.Timer(
                    self.batch_timeout, 
                    self._flush_batch
                )
                self._batch_timer.start()

import functools
import time

def cached_discovery(ttl_seconds=5.0):
    """Decorator to cache discovery results"""
    cache = {'result': None, 'timestamp': 0}
    
    @functools.wraps(discover_channels)
    def wrapper(*args, **kwargs):
        now = time.time()
        if cache['result'] and (now - cache['timestamp']) < ttl_seconds:
            logger.debug(f"Using cached discovery (age: {now - cache['timestamp']:.1f}s)")
            return cache['result']
        
        result = discover_channels(*args, **kwargs)
        cache['result'] = result
        cache['timestamp'] = now
        return result
    
    return wrapper

# Usage
discover_channels_cached = cached_discovery(ttl_seconds=10.0)

from typing import Callable, Dict
import threading

class EventDrivenRadiodControl(RadiodControl):
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._callbacks: Dict[str, list] = {
            'channel_created': [],
            'frequency_changed': [],
            'status_update': [],
        }
        self._listener_thread = None
        self._running = False
    
    def on(self, event: str, callback: Callable):
        """Register event callback"""
        if event not in self._callbacks:
            self._callbacks[event] = []
        self._callbacks[event].append(callback)
    
    def start_listening(self):
        """Start background status listener"""
        self._running = True
        self._listener_thread = threading.Thread(
            target=self._listen_loop,
            daemon=True
        )
        self._listener_thread.start()
    
    def _listen_loop(self):
        """Background listener for status updates"""
        status_sock = self._setup_status_listener()
        while self._running:
            try:
                buffer, _ = status_sock.recvfrom(8192)
                status = self._decode_status_response(buffer)
                
                # Trigger callbacks
                for callback in self._callbacks['status_update']:
                    callback(status)
            except socket.timeout:
                continue

class WSPRMonitor(RadiodControl):
    """High-level interface for WSPR monitoring"""
    WSPR_BANDS = {
        '160m': 1.8366e6,
        '80m': 3.5686e6,
        '40m': 7.0386e6,
        '30m': 10.1387e6,
        '20m': 14.0956e6,
    }
    
    def monitor_all_bands(self):
        """Create channels for all WSPR bands"""
        for band, freq in self.WSPR_BANDS.items():
            self.create_channel(
                ssrc=int(freq),
                frequency_hz=freq,
                preset='usb',
                sample_rate=12000
            )
            logger.info(f"Monitoring {band} WSPR")

class SatelliteTracker(RadiodControl):
    """High-level interface for satellite tracking with Doppler"""
    def track_satellite(self, satellite_name: str, observer_location: tuple):
        """Track satellite with automatic Doppler correction"""
        # Integration with orbital prediction libraries
        pass

import socket
import struct

class RTPReceiver:
    """Receive RTP audio/IQ streams from radiod"""
    def __init__(self, multicast_addr: str, port: int):
        self.addr = multicast_addr
        self.port = port
        self.sock = None
    
    def start(self):
        """Start receiving RTP packets"""
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.sock.bind(('0.0.0.0', self.port))
        
        # Join multicast group
        mreq = struct.pack('4sl', 
                          socket.inet_aton(self.addr),
                          socket.INADDR_ANY)
        self.sock.setsockopt(socket.IPPROTO_IP, 
                            socket.IP_ADD_MEMBERSHIP, 
                            mreq)
    
    def receive_samples(self, num_samples: int = 1024):
        """Receive IQ samples from RTP stream"""
        # Parse RTP header, extract samples
        pass

# Integration with RadiodControl
class RadiodWithRTP(RadiodControl):
    def create_channel_with_receiver(self, ssrc, frequency_hz, **kwargs):
        """Create channel and return RTP receiver"""
        # Discover multicast address for this channel
        self.create_channel(ssrc, frequency_hz, **kwargs)
        channels = discover_channels(self.status_address)
        
        if ssrc in channels:
            channel = channels[ssrc]
            rtp = RTPReceiver(
                channel.multicast_address,
                channel.port
            )
            rtp.start()
            return rtp

class RadiodPool:
    """Connection pool for multiple radiod instances"""
    def __init__(self):
        self._connections: Dict[str, RadiodControl] = {}
        self._lock = threading.Lock()
    
    def get_connection(self, status_address: str) -> RadiodControl:
        """Get or create connection to radiod instance"""
        with self._lock:
            if status_address not in self._connections:
                self._connections[status_address] = RadiodControl(status_address)
            return self._connections[status_address]
    
    def close_all(self):
        """Close all connections"""
        for conn in self._connections.values():
            conn.close()

class AutoRadiodControl:
    """Automatically discover and connect to radiod"""
    @classmethod
    def auto_connect(cls, service_name: str = None):
        """Discover and connect to radiod automatically"""
        services = discover_radiod_services()
        
        if not services:
            raise DiscoveryError("No radiod services found on network")
        
        if service_name:
            # Find specific service
            for svc in services:
                if service_name in svc['name']:
                    return cls(svc['address'])
            raise DiscoveryError(f"Service {service_name} not found")
        else:
            # Use first available
            return cls(services[0]['address'])

# Usage:
control = AutoRadiodControl.auto_connect()
# OR
control = AutoRadiodControl.auto_connect(service_name='radiod@hf')

# tests/test_security.py
def test_malformed_packets():
    """Test handling of malformed status packets"""
    control = RadiodControl("test.local")
    
    malformed_packets = [
        b'',  # Empty
        b'\x00',  # Only type byte
        b'\x00\x21\x08',  # Incomplete
        b'\x00\x21\xFF' + b'\x00' * 300,  # Invalid length
    ]
    
    for packet in malformed_packets:
        status = control._decode_status_response(packet)
        # Should not crash, should return empty or minimal dict

def test_concurrent_commands():
    """Test thread safety"""
    control = RadiodControl("test.local")
    errors = []
    
    def worker():
        try:
            for i in range(100):
                control.set_frequency(ssrc=10000, frequency_hz=14.0e6 + i)
        except Exception as e:
            errors.append(e)
    
    threads = [threading.Thread(target=worker) for _ in range(10)]
    for t in threads:
        t.start()
    for t in threads:
        t.join()
    
    assert len(errors) == 0, f"Concurrent errors: {errors}"