diff options
| author | gabrix73 <gabriel1@frozenstar.info> | 2026-04-10 16:10:48 +0200 |
|---|---|---|
| committer | gabrix73 <gabriel1@frozenstar.info> | 2026-04-10 16:10:48 +0200 |
| commit | 88375da30ea667ec25c2000874557fcc4785a558 (patch) | |
| tree | 71364a43230977481c6d3280246aa8bb48562bdc /transport/mesh/storeforward.go | |
| download | khimera-88375da30ea667ec25c2000874557fcc4785a558.tar.gz khimera-88375da30ea667ec25c2000874557fcc4785a558.tar.xz khimera-88375da30ea667ec25c2000874557fcc4785a558.zip | |
Initial commit: Veilith v0.9 - P2P Encrypted Messenger
Military-grade secure messenger with:
- End-to-End Encryption (Noise Protocol XX)
- Forward Secrecy (automatic session rotation)
- Anti-Traffic Analysis (padding + cover traffic)
- Secure Deletion (DOD 5220.22-M)
- Tor Integration (embedded with bridge support)
- Portable Mode (no installation required)
Generated with Claude Code
Diffstat (limited to 'transport/mesh/storeforward.go')
| -rw-r--r-- | transport/mesh/storeforward.go | 319 |
1 files changed, 319 insertions, 0 deletions
diff --git a/transport/mesh/storeforward.go b/transport/mesh/storeforward.go new file mode 100644 index 0000000..1fcb93d --- /dev/null +++ b/transport/mesh/storeforward.go @@ -0,0 +1,319 @@ +// Package mesh - Store-and-forward messaging for offline scenarios +package mesh + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "sync" + "time" +) + +// MessageQueue implements store-and-forward messaging +// Critical for military scenarios where connectivity is intermittent +type MessageQueue struct { + pending map[string][]*QueuedMessage // recipient -> messages + delivered map[string]bool // message ID -> delivered + queueDir string // Persistent storage directory + maxAge time.Duration // Auto-delete after this time + maxSize int // Max messages in queue + encrypted bool // Encrypt queue on disk + mu sync.RWMutex + onDelivered func(string) // Callback when message delivered +} + +// QueuedMessage represents a message waiting for delivery +type QueuedMessage struct { + ID string `json:"id"` // Unique message ID + From string `json:"from"` // Sender public key + To string `json:"to"` // Recipient public key + Content []byte `json:"content"` // Encrypted message content + Timestamp time.Time `json:"timestamp"` // When queued + Attempts int `json:"attempts"` // Delivery attempts + MaxAttempts int `json:"max_attempts"` // Give up after this many + Priority int `json:"priority"` // Higher = more important + Hops int `json:"hops"` // How many hops so far + MaxHops int `json:"max_hops"` // Don't forward beyond this +} + +// QueueConfig holds message queue configuration +type QueueConfig struct { + QueueDir string // Where to store pending messages + MaxAge time.Duration // Auto-delete messages older than this + MaxSize int // Max messages per recipient + MaxAttempts int // Max delivery attempts + MaxHops int // Max forwarding hops + Encrypted bool // Encrypt queue on disk +} + +// NewMessageQueue creates a new message queue +func NewMessageQueue(config *QueueConfig) (*MessageQueue, error) { + if config == nil { + config = &QueueConfig{ + QueueDir: filepath.Join(os.TempDir(), "veilith-queue"), + MaxAge: 24 * time.Hour, + MaxSize: 1000, + MaxAttempts: 10, + MaxHops: 5, + Encrypted: true, + } + } + + // Create queue directory + if err := os.MkdirAll(config.QueueDir, 0700); err != nil { + return nil, fmt.Errorf("failed to create queue dir: %w", err) + } + + mq := &MessageQueue{ + pending: make(map[string][]*QueuedMessage), + delivered: make(map[string]bool), + queueDir: config.QueueDir, + maxAge: config.MaxAge, + maxSize: config.MaxSize, + encrypted: config.Encrypted, + } + + // Load existing queue from disk + if err := mq.loadFromDisk(); err != nil { + return nil, fmt.Errorf("failed to load queue: %w", err) + } + + // Start background cleanup + go mq.cleanupLoop() + + return mq, nil +} + +// Enqueue adds a message to the queue +func (mq *MessageQueue) Enqueue(msg *QueuedMessage) error { + mq.mu.Lock() + defer mq.mu.Unlock() + + // Check if already delivered + if mq.delivered[msg.ID] { + return fmt.Errorf("message already delivered") + } + + // Check queue size limit + if len(mq.pending[msg.To]) >= mq.maxSize { + return fmt.Errorf("queue full for recipient %s", msg.To) + } + + // Add to pending queue + mq.pending[msg.To] = append(mq.pending[msg.To], msg) + + // Persist to disk + if err := mq.saveToDisk(); err != nil { + return fmt.Errorf("failed to save queue: %w", err) + } + + return nil +} + +// Dequeue retrieves messages for a specific recipient +func (mq *MessageQueue) Dequeue(recipient string) []*QueuedMessage { + mq.mu.Lock() + defer mq.mu.Unlock() + + messages := mq.pending[recipient] + delete(mq.pending, recipient) + + // Mark as delivered + for _, msg := range messages { + mq.delivered[msg.ID] = true + if mq.onDelivered != nil { + go mq.onDelivered(msg.ID) + } + } + + // Persist to disk + mq.saveToDisk() + + return messages +} + +// GetPendingCount returns number of pending messages for a recipient +func (mq *MessageQueue) GetPendingCount(recipient string) int { + mq.mu.RLock() + defer mq.mu.RUnlock() + return len(mq.pending[recipient]) +} + +// GetTotalPendingCount returns total number of pending messages +func (mq *MessageQueue) GetTotalPendingCount() int { + mq.mu.RLock() + defer mq.mu.RUnlock() + + total := 0 + for _, messages := range mq.pending { + total += len(messages) + } + return total +} + +// GetPendingRecipients returns list of recipients with pending messages +func (mq *MessageQueue) GetPendingRecipients() []string { + mq.mu.RLock() + defer mq.mu.RUnlock() + + recipients := make([]string, 0, len(mq.pending)) + for recipient := range mq.pending { + recipients = append(recipients, recipient) + } + return recipients +} + +// IncrementAttempts increments delivery attempt counter +func (mq *MessageQueue) IncrementAttempts(msgID string) { + mq.mu.Lock() + defer mq.mu.Unlock() + + for _, messages := range mq.pending { + for _, msg := range messages { + if msg.ID == msgID { + msg.Attempts++ + mq.saveToDisk() + return + } + } + } +} + +// ShouldForward checks if message should be forwarded (multi-hop routing) +func (mq *MessageQueue) ShouldForward(msg *QueuedMessage) bool { + return msg.Hops < msg.MaxHops && msg.Attempts < msg.MaxAttempts +} + +// saveToDisk persists the queue to disk +func (mq *MessageQueue) saveToDisk() error { + queuePath := filepath.Join(mq.queueDir, "queue.json") + + // Serialize queue + data, err := json.MarshalIndent(mq.pending, "", " ") + if err != nil { + return fmt.Errorf("failed to marshal queue: %w", err) + } + + // TODO: Encrypt if mq.encrypted is true + // For now, just save plaintext + + // Write to temp file first (atomic write) + tempPath := queuePath + ".tmp" + if err := os.WriteFile(tempPath, data, 0600); err != nil { + return fmt.Errorf("failed to write queue: %w", err) + } + + // Atomic rename + if err := os.Rename(tempPath, queuePath); err != nil { + return fmt.Errorf("failed to rename queue: %w", err) + } + + return nil +} + +// loadFromDisk loads the queue from disk +func (mq *MessageQueue) loadFromDisk() error { + queuePath := filepath.Join(mq.queueDir, "queue.json") + + // Check if file exists + if _, err := os.Stat(queuePath); os.IsNotExist(err) { + // No existing queue - that's fine + return nil + } + + // Read file + data, err := os.ReadFile(queuePath) + if err != nil { + return fmt.Errorf("failed to read queue: %w", err) + } + + // TODO: Decrypt if mq.encrypted is true + + // Parse JSON + if err := json.Unmarshal(data, &mq.pending); err != nil { + return fmt.Errorf("failed to unmarshal queue: %w", err) + } + + return nil +} + +// cleanupLoop periodically removes old messages +func (mq *MessageQueue) cleanupLoop() { + ticker := time.NewTicker(10 * time.Minute) + defer ticker.Stop() + + for range ticker.C { + mq.cleanup() + } +} + +// cleanup removes expired messages +func (mq *MessageQueue) cleanup() { + mq.mu.Lock() + defer mq.mu.Unlock() + + now := time.Now() + removed := 0 + + for recipient, messages := range mq.pending { + // Filter out expired messages + validMessages := make([]*QueuedMessage, 0) + for _, msg := range messages { + // Remove if too old or too many attempts + if now.Sub(msg.Timestamp) > mq.maxAge || msg.Attempts >= msg.MaxAttempts { + removed++ + continue + } + validMessages = append(validMessages, msg) + } + + if len(validMessages) > 0 { + mq.pending[recipient] = validMessages + } else { + delete(mq.pending, recipient) + } + } + + // Clean delivered map (keep only recent entries) + for msgID := range mq.delivered { + // TODO: Add timestamp to delivered map to clean old entries + _ = msgID + } + + if removed > 0 { + mq.saveToDisk() + } +} + +// Clear removes all messages from queue +func (mq *MessageQueue) Clear() error { + mq.mu.Lock() + defer mq.mu.Unlock() + + mq.pending = make(map[string][]*QueuedMessage) + mq.delivered = make(map[string]bool) + + return mq.saveToDisk() +} + +// SetOnDelivered sets callback for when message is delivered +func (mq *MessageQueue) SetOnDelivered(callback func(string)) { + mq.mu.Lock() + defer mq.mu.Unlock() + mq.onDelivered = callback +} + +// String returns a string representation +func (mq *MessageQueue) String() string { + mq.mu.RLock() + defer mq.mu.RUnlock() + + totalPending := 0 + for _, messages := range mq.pending { + totalPending += len(messages) + } + + return fmt.Sprintf("MessageQueue(pending=%d, delivered=%d, recipients=%d)", + totalPending, len(mq.delivered), len(mq.pending)) +} |
