KrakenHashes Agent Development Guide¶

// From cmd/agent/main.go
func main() {
    // 1. Initialize debug logging
    debug.Reinitialize()

    // 2. Load configuration from .env
    cfg := loadConfig()

    // 3. Initialize data directories
    dataDirs, err := config.GetDataDirs()

    // 4. Create metrics collector
    collector, err := metrics.New(metrics.Config{
        CollectionInterval: time.Duration(cfg.heartbeatInterval) * time.Second,
        EnableGPU:          true,
    })

    // 5. Load or register agent credentials
    agentID, cert, err := agent.LoadCredentials()

    // 6. Create WebSocket connection
    conn, err := agent.NewConnection(urlConfig)

    // 7. Create job manager with hardware monitor
    hwMonitor := conn.GetHardwareMonitor()
    jobManager = jobs.NewJobManager(agentConfig, nil, hwMonitor)

    // 8. Start connection and maintenance
    conn.Start()

    // 9. Detect and send device information
    conn.DetectAndSendDevices()
}

// From internal/hardware/hashcat_detector.go
type HashcatDetector struct {
    binaryPath     string
    dataDirectory  string
}

func (d *HashcatDetector) DetectDevices() (*types.DeviceDetectionResult, error) {
    // Build hashcat command with device info flags
    args := []string{"-I", "--machine-readable", "--quiet"}

    cmd := exec.CommandContext(ctx, d.binaryPath, args...)
    output, err := cmd.Output()

    // Parse hashcat output
    devices, backends, err := d.parseHashcatOutput(string(output))

    return &types.DeviceDetectionResult{
        Devices:          devices,
        OpenCLBackends:   backends,
        DetectedAt:       time.Now(),
        HashcatVersion:   d.getHashcatVersion(),
    }, nil
}

// From internal/hardware/types/device.go
type Device struct {
    ID          int    `json:"id"`
    Type        string `json:"type"`        // "gpu" or "cpu"
    Brand       string `json:"brand"`       // "nvidia", "amd", "intel"
    Name        string `json:"name"`
    Processor   string `json:"processor"`
    Memory      int64  `json:"memory"`      // In bytes
    DriverVersion string `json:"driver_version"`

    // Performance characteristics
    PCIeBus     string `json:"pcie_bus"`
    CoreCount   int    `json:"core_count"`
    ClockSpeed  int    `json:"clock_speed"` // In MHz

    // Runtime state
    Enabled     bool   `json:"enabled"`
}

// From internal/hardware/monitor.go
type Monitor struct {
    mu             sync.RWMutex
    devices        []types.Device
    hashcatDetector *HashcatDetector
}

// Detect devices using hashcat
func (m *Monitor) DetectDevices() (*types.DeviceDetectionResult, error) {
    result, err := m.hashcatDetector.DetectDevices()
    if err != nil {
        return nil, err
    }

    // Store devices in monitor
    m.mu.Lock()
    m.devices = result.Devices
    m.mu.Unlock()

    return result, nil
}

// Get enabled device flags for hashcat (-d parameter)
func (m *Monitor) GetEnabledDeviceFlags() string {
    m.mu.RLock()
    defer m.mu.RUnlock()

    return BuildDeviceFlags(m.devices)
}

// From internal/jobs/jobs.go
type JobManager struct {
    executor         *HashcatExecutor
    config           *config.Config
    progressCallback func(*JobProgress)
    outputCallback   func(taskID string, output string, isError bool)
    fileSync         *filesync.FileSync
    hwMonitor        HardwareMonitor

    mutex           sync.RWMutex
    activeJobs      map[string]*JobExecution
}

// Process a job assignment from the backend
func (jm *JobManager) ProcessJobAssignment(ctx context.Context, assignmentData []byte) error {
    var assignment JobTaskAssignment
    err := json.Unmarshal(assignmentData, &assignment)

    // 1. Ensure hashlist is available
    err = jm.ensureHashlist(ctx, &assignment)

    // 2. Ensure rule chunks are available if needed
    err = jm.ensureRuleChunks(ctx, &assignment)

    // 3. Start job execution
    process, err := jm.executor.ExecuteTask(ctx, &assignment)

    // 4. Monitor job progress
    go jm.monitorJobProgress(ctx, jobExecution)

    return nil
}

// From internal/jobs/hashcat_executor.go
type HashcatExecutor struct {
    dataDirectory      string
    activeProcesses    map[string]*HashcatProcess
    mu                 sync.RWMutex
    agentExtraParams   string
    deviceFlagsCallback func() string
}

// Execute a hashcat task
func (e *HashcatExecutor) ExecuteTask(ctx context.Context, assignment *JobTaskAssignment) (*HashcatProcess, error) {
    // 1. Build hashcat command
    args := e.buildHashcatCommand(assignment)

    // 2. Create process with output capture
    cmd := exec.CommandContext(ctx, binaryPath, args...)

    // 3. Set up output pipes
    stdout, _ := cmd.StdoutPipe()
    stderr, _ := cmd.StderrPipe()

    // 4. Start the process
    err := cmd.Start()

    // 5. Monitor output and parse progress
    go e.monitorOutput(process, stdout, stderr)

    return process, nil
}

// Build hashcat command with all parameters
func (e *HashcatExecutor) buildHashcatCommand(assignment *JobTaskAssignment) []string {
    args := []string{
        "-m", strconv.Itoa(assignment.HashType),
        "-a", strconv.Itoa(assignment.AttackMode),
        "--status",
        "--status-json",
        "--status-timer", strconv.Itoa(assignment.ReportInterval),
    }

    // Add device flags if callback is set
    if e.deviceFlagsCallback != nil {
        deviceFlags := e.deviceFlagsCallback()
        if deviceFlags != "" {
            args = append(args, "-d", deviceFlags)
        }
    }

    // Add skip and limit for keyspace splitting
    if assignment.KeyspaceStart > 0 {
        args = append(args, "-s", strconv.FormatInt(assignment.KeyspaceStart, 10))
    }
    if assignment.KeyspaceEnd > assignment.KeyspaceStart {
        limit := assignment.KeyspaceEnd - assignment.KeyspaceStart
        args = append(args, "-l", strconv.FormatInt(limit, 10))
    }

    return args
}

// From internal/jobs/types.go
type JobProgress struct {
    TaskID            string         `json:"task_id"`
    Status            string         `json:"status"`
    Progress          float64        `json:"progress"`
    Speed             int64          `json:"speed"`
    DeviceSpeeds      []DeviceSpeed  `json:"device_speeds"`
    TimeRemaining     int            `json:"time_remaining"`
    KeyspaceProcessed int64          `json:"keyspace_processed"`
    CrackedCount      int            `json:"cracked_count"`
    CrackedHashes     []CrackedHash  `json:"cracked_hashes,omitempty"`
    ErrorMessage      string         `json:"error_message,omitempty"`
}

// From internal/jobs/state.go
type TaskState int

const (
    TaskStateIdle       TaskState = iota  // No task running
    TaskStateRunning                      // Task actively executing
    TaskStateCompleting                   // Task done, waiting for backend ACK
    TaskStateStopped                      // Task stopped by user
    TaskStateFailed                       // Task failed
)

// From internal/jobs/state.go
type TaskStateManager struct {
    mu                sync.RWMutex
    currentState      TaskState
    currentTaskID     string
    stateChangedAt    time.Time
    completionPending bool
    pendingTaskID     string
}

// State transition methods
func (m *TaskStateManager) TransitionTo(state TaskState, taskID string)
func (m *TaskStateManager) GetState() (TaskState, string)
func (m *TaskStateManager) GetStateInfo() (TaskState, string, time.Time)
func (m *TaskStateManager) SetCompletionPending(taskID string)
func (m *TaskStateManager) GetCompletionPending() (bool, string)
func (m *TaskStateManager) TransitionToIdle()

// From internal/jobs/jobs.go
func (jm *JobManager) SetAckWaitCallback(callback func(taskID string, resendFunc func() error) bool) {
    jm.ackWaitCallback = callback
}

// After task completion
if jm.ackWaitCallback != nil {
    jm.stateManager.TransitionTo(TaskStateCompleting, taskID)

    // Create resend function for retries
    resendFunc := func() error {
        return jm.sendCompletionProgress(taskID, finalProgress)
    }

    // Wait for ACK (blocks until ACK received or timeout)
    ackReceived := jm.ackWaitCallback(taskID, resendFunc)

    if !ackReceived {
        // Mark as pending for state sync recovery
        jm.stateManager.SetCompletionPending(taskID)
    }
}

jm.stateManager.TransitionToIdle()

// Sent by backend
type TaskCompleteAck struct {
    TaskID    string    `json:"task_id"`
    Timestamp time.Time `json:"timestamp"`  // For duplicate detection
    Success   bool      `json:"success"`
    Message   string    `json:"message,omitempty"`
}

// From internal/agent/connection.go
const (
    AckWaitTimeout    = 30 * time.Second  // Timeout per ACK attempt
    AckMaxRetries     = 3                  // Maximum retry attempts
)

const (
    StuckDetectionTimeout = 2 * time.Minute  // Max time in COMPLETING state
    StuckCheckInterval    = 30 * time.Second // Check frequency
)

// From internal/agent/connection.go
func (c *Connection) StartStuckDetection(ctx context.Context) {
    ticker := time.NewTicker(StuckCheckInterval)
    defer ticker.Stop()

    for {
        select {
        case <-ctx.Done():
            return
        case <-ticker.C:
            c.checkForStuckState()
        }
    }
}

func (c *Connection) checkForStuckState() {
    state, taskID, changedAt := c.jobManager.GetStateInfo()

    if state == TaskStateCompleting {
        if time.Since(changedAt) > StuckDetectionTimeout {
            debug.Warning("Stuck detection: Agent in COMPLETING state for %v, forcing recovery",
                time.Since(changedAt))
            c.forceRecovery(taskID)
        }
    }
}

func (c *Connection) forceRecovery(taskID string) {
    // Mark completion as pending for state sync
    c.jobManager.SetCompletionPending(taskID)

    // Force transition to idle
    c.jobManager.TransitionToIdle()

    debug.Info("Force recovery complete, agent now idle")
}

// From internal/jobs/jobs.go
func (jm *JobManager) cleanupCompletedTask(taskID string) {
    jm.mutex.Lock()
    defer jm.mutex.Unlock()

    // 1. Remove from active jobs map
    if job, exists := jm.activeJobs[taskID]; exists {
        // Cache completion info for later queries
        jm.completedTaskCache[taskID] = &CompletedTaskInfo{
            CompletedAt: time.Now(),
            FinalProgress: job.FinalProgress,
        }
        delete(jm.activeJobs, taskID)
    }

    // 2. Clean up attack files
    jm.cleanupAttackFiles(taskID)
}

// Called synchronously before logging completion
func (jm *JobManager) handleTaskCompletion(taskID string, progress *JobProgress) {
    // FIRST: Synchronous cleanup
    jm.cleanupCompletedTask(taskID)

    // THEN: Log completion (state is already clean)
    debug.Info("Task %s completed successfully", taskID)

    // THEN: Wait for ACK
    if jm.ackWaitCallback != nil {
        // ... ACK waiting logic
    }
}

// From internal/jobs/jobs.go
func (jm *JobManager) calculateProgress(hashcatStatus map[string]interface{}) float64 {
    // Prefer effective progress values
    effectiveProgress := getInt64(hashcatStatus, "effective_progress")
    totalEffectiveKeyspace := getInt64(hashcatStatus, "total_effective_keyspace")

    if effectiveProgress > 0 && totalEffectiveKeyspace > 0 {
        return float64(effectiveProgress) / float64(totalEffectiveKeyspace) * 100.0
    }

    // Fallback to base keyspace
    progress := hashcatStatus["progress"].([]interface{})
    return float64(progress[0].(int64)) / float64(progress[1].(int64)) * 100.0
}

// From internal/agent/connection.go
type Connection struct {
    ws              *websocket.Conn
    urlConfig       *config.URLConfig
    hwMonitor       *hardware.Monitor
    outbound        chan *WSMessage
    done            chan struct{}
    isConnected     atomic.Bool
    tlsConfig       *tls.Config
    fileSync        *filesync.FileSync
    jobManager      JobManager
}

// WebSocket message types
type WSMessageType string

const (
    WSTypeHardwareInfo         WSMessageType = "hardware_info"
    WSTypeMetrics              WSMessageType = "metrics"
    WSTypeHeartbeat            WSMessageType = "heartbeat"
    WSTypeAgentStatus          WSMessageType = "agent_status"
    WSTypeFileSyncRequest      WSMessageType = "file_sync_request"
    WSTypeFileSyncResponse     WSMessageType = "file_sync_response"
    WSTypeTaskAssignment       WSMessageType = "task_assignment"
    WSTypeJobProgress          WSMessageType = "job_progress"
    WSTypeJobStop              WSMessageType = "job_stop"
    WSTypeBenchmarkRequest     WSMessageType = "benchmark_request"
    WSTypeBenchmarkResult      WSMessageType = "benchmark_result"
    WSTypeHashcatOutput        WSMessageType = "hashcat_output"
    WSTypeDeviceDetection      WSMessageType = "device_detection"
    WSTypeDeviceUpdate         WSMessageType = "device_update"
    WSTypeCrackBatch           WSMessageType = "crack_batch"
    WSTypeCrackBatchesComplete WSMessageType = "crack_batches_complete"  // NEW in v1.3.1
)

// From internal/agent/connection.go - readPump method
func (c *Connection) readPump() {
    for {
        var msg WSMessage
        err := c.ws.ReadJSON(&msg)

        switch msg.Type {
        case WSTypeTaskAssignment:
            // Process job assignment
            if err := c.jobManager.ProcessJobAssignment(ctx, msg.Payload); err != nil {
                debug.Error("Failed to process job assignment: %v", err)
            }

        case WSTypeFileSyncRequest:
            // Handle file sync request
            go c.handleFileSyncAsync(requestPayload)

        case WSTypeBenchmarkRequest:
            // Run speed test
            go func() {
                totalSpeed, deviceSpeeds, err := executor.RunSpeedTest(ctx, assignment, testDuration)
                // Send results back
            }()

        case WSTypeDeviceUpdate:
            // Update device enabled/disabled state
            c.hwMonitor.UpdateDeviceStatus(updatePayload.DeviceID, updatePayload.Enabled)
        }
    }
}

// Send job progress to backend
func (c *Connection) SendJobProgress(progress *jobs.JobProgress) error {
    progressJSON, err := json.Marshal(progress)

    msg := &WSMessage{
        Type:      WSTypeJobProgress,
        Payload:   progressJSON,
        Timestamp: time.Now(),
    }

    select {
    case c.outbound <- msg:
        return nil
    case <-time.After(5 * time.Second):
        return fmt.Errorf("failed to queue job progress: channel blocked")
    }
}

// Send final progress with crack count
progress := &JobProgress{
    TaskID:          taskID,
    Status:          "completed",
    ProgressPercent: 100.0,
    CrackedCount:    totalCracks,  // Expected number of cracks
}
c.SendJobProgress(progress)

// Send crack batches (batched during execution)
for _, batch := range crackBatches {
    crackBatch := &CrackBatch{
        TaskID:        taskID,
        CrackedHashes: batch,
    }
    c.SendCrackBatch(crackBatch)
}

// Signal all batches sent
completion := &CrackBatchesComplete{
    TaskID: taskID,
}
c.SendCrackBatchesComplete(completion)

// Final progress message
type JobProgress struct {
    TaskID          string  `json:"task_id"`
    Status          string  `json:"status"`           // "completed"
    ProgressPercent float64 `json:"progress_percent"` // 100.0
    CrackedCount    int     `json:"cracked_count"`    // Expected cracks
    // ... other fields
}

// Crack batch message
type CrackBatch struct {
    TaskID        string         `json:"task_id"`
    CrackedHashes []CrackedHash  `json:"cracked_hashes"`
}

type CrackedHash struct {
    Hash         string `json:"hash"`
    Plaintext    string `json:"plaintext"`
    OriginalLine string `json:"original_line"`
}

// Completion signal message
type CrackBatchesComplete struct {
    TaskID string `json:"task_id"`
}

// From internal/sync/sync.go
type FileSync struct {
    client     *http.Client
    urlConfig  *config.URLConfig
    dataDirs   *config.DataDirs
    sem        chan struct{} // Semaphore for concurrent downloads
    maxRetries int
    apiKey     string
    agentID    string
}

// Download a file from the backend
func (fs *FileSync) DownloadFileFromInfo(ctx context.Context, fileInfo *FileInfo) error {
    // 1. Check if file already exists with correct hash
    if fs.fileExistsWithHash(fileInfo) {
        return nil
    }

    // 2. Build download URL
    url := fs.buildDownloadURL(fileInfo)

    // 3. Create authenticated request
    req, _ := http.NewRequestWithContext(ctx, "GET", url, nil)
    req.Header.Set("X-API-Key", fs.apiKey)
    req.Header.Set("X-Agent-ID", fs.agentID)

    // 4. Download to temporary file
    tempPath := finalPath + ".tmp"
    err := fs.downloadToFile(req, tempPath)

    // 5. Verify hash
    if !fs.verifyFileHash(tempPath, fileInfo.MD5Hash) {
        return fmt.Errorf("hash mismatch")
    }

    // 6. Move to final location
    os.Rename(tempPath, finalPath)

    // 7. Extract if it's a binary archive
    if fileInfo.FileType == "binary" && strings.HasSuffix(fileInfo.Name, ".7z") {
        fs.ExtractBinary7z(finalPath, targetDir)
    }

    return nil
}

// From internal/config/dirs.go
type DataDirs struct {
    Binaries   string // /path/to/data/binaries
    Wordlists  string // /path/to/data/wordlists
    Rules      string // /path/to/data/rules
    Hashlists  string // /path/to/data/hashlists
}

// Wordlist categories:
// - general/     # Common wordlists
// - specialized/ # Domain-specific lists
// - targeted/    # Custom lists for specific targets
// - custom/      # User-uploaded lists

// Rule categories:
// - hashcat/     # Hashcat rule files
// - john/        # John the Ripper rules
// - custom/      # User-created rules
// - chunks/      # Split rule files for distributed processing

// From internal/metrics/collector.go
type Collector struct {
    interval   time.Duration
    gpuEnabled bool
}

// Collect system metrics
func (c *Collector) Collect() (*SystemMetrics, error) {
    metrics := &SystemMetrics{}

    // CPU metrics using gopsutil
    percentage, _ := cpu.Percent(time.Second, false)
    metrics.CPUUsage = percentage[0]

    // Memory metrics
    vmem, _ := mem.VirtualMemory()
    metrics.MemoryUsage = vmem.UsedPercent

    // GPU metrics come from hashcat JSON status
    // during job execution

    return metrics, nil
}

// From internal/agent/connection.go
type MetricsData struct {
    AgentID     int               `json:"agent_id"`
    CollectedAt time.Time         `json:"collected_at"`
    CPUs        []CPUMetrics      `json:"cpus"`
    GPUs        []GPUMetrics      `json:"gpus"`
    Memory      MemoryMetrics     `json:"memory"`
    Disk        []DiskMetrics     `json:"disk"`
    Network     []NetworkMetrics  `json:"network"`
    Process     []ProcessMetrics  `json:"process"`
}

// 1. Define the message type in connection.go
const WSTypeNewFeature WSMessageType = "new_feature"

// 2. Create payload structures
type NewFeatureRequest struct {
    Parameter1 string `json:"parameter1"`
    Parameter2 int    `json:"parameter2"`
}

// 3. Add handler in readPump
case WSTypeNewFeature:
    var payload NewFeatureRequest
    if err := json.Unmarshal(msg.Payload, &payload); err != nil {
        debug.Error("Failed to parse new feature: %v", err)
        continue
    }

    // Handle the feature
    go c.handleNewFeature(payload)

// 4. Implement the handler
func (c *Connection) handleNewFeature(payload NewFeatureRequest) {
    // Implementation
}

// 1. Update device detection in hashcat_detector.go
func (d *HashcatDetector) parseHashcatOutput(output string) ([]types.Device, []types.OpenCLBackend, error) {
    // Add parsing for new hardware types

    // Example: Detect new GPU vendor
    if strings.Contains(line, "NewVendor") {
        device.Brand = "newvendor"
        device.Type = "gpu"
    }
}

// 2. Add device-specific handling
func BuildDeviceFlags(devices []types.Device) string {
    // Add logic for new device types
}

// 1. Update JobTaskAssignment structure
type JobTaskAssignment struct {
    // Existing fields...

    NewFeature string `json:"new_feature"`
}

// 2. Update hashcat command building
func (e *HashcatExecutor) buildHashcatCommand(assignment *JobTaskAssignment) []string {
    // Add new hashcat parameters
    if assignment.NewFeature != "" {
        args = append(args, "--new-feature", assignment.NewFeature)
    }
}

// 3. Update progress monitoring if needed
func (e *HashcatExecutor) parseHashcatStatus(status map[string]interface{}) *JobProgress {
    // Parse new status fields
}

// Example from agent_test.go
func TestGetAgentID(t *testing.T) {
    tests := []struct {
        name        string
        setupFunc   func(configDir string) error
        expectedID  int
        wantErr     bool
    }{
        {
            name: "successful ID retrieval",
            setupFunc: func(configDir string) error {
                return auth.SaveAgentKey(configDir, "test-api-key", "456")
            },
            expectedID: 456,
            wantErr:    false,
        },
    }

    for _, tt := range tests {
        t.Run(tt.name, func(t *testing.T) {
            tempDir := t.TempDir()
            t.Setenv("KH_CONFIG_DIR", tempDir)

            if tt.setupFunc != nil {
                err := tt.setupFunc(tempDir)
                require.NoError(t, err)
            }

            id, err := GetAgentID()

            if tt.wantErr {
                assert.Error(t, err)
            } else {
                assert.NoError(t, err)
                assert.Equal(t, tt.expectedID, id)
            }
        })
    }
}

// Test WebSocket connection
func TestWebSocketConnection(t *testing.T) {
    // Create test server
    server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        upgrader := websocket.Upgrader{}
        conn, _ := upgrader.Upgrade(w, r, nil)
        defer conn.Close()

        // Test message exchange
        var msg WSMessage
        conn.ReadJSON(&msg)
        assert.Equal(t, WSTypeHeartbeat, msg.Type)

        // Send response
        conn.WriteJSON(WSMessage{
            Type: WSTypeHeartbeat,
            Timestamp: time.Now(),
        })
    }))
    defer server.Close()

    // Test connection
    conn, err := NewConnection(urlConfig)
    assert.NoError(t, err)

    err = conn.Start()
    assert.NoError(t, err)
}

// From internal/mocks/
type MockJobManager struct {
    mock.Mock
}

func (m *MockJobManager) ProcessJobAssignment(ctx context.Context, data []byte) error {
    args := m.Called(ctx, data)
    return args.Error(0)
}

// Use in tests
func TestJobProcessing(t *testing.T) {
    mockJM := new(MockJobManager)
    mockJM.On("ProcessJobAssignment", mock.Anything, mock.Anything).Return(nil)

    conn := &Connection{
        jobManager: mockJM,
    }

    // Test job processing
    msg := WSMessage{
        Type: WSTypeTaskAssignment,
        Payload: json.RawMessage(`{"task_id": "test-123"}`),
    }

    // Process message
    // Assert expectations
    mockJM.AssertExpectations(t)
}

func BenchmarkHashcatCommandBuilding(b *testing.B) {
    executor := NewHashcatExecutor("/data")
    assignment := &JobTaskAssignment{
        TaskID:     "bench-task",
        HashType:   0,
        AttackMode: 0,
        WordlistPaths: []string{"wordlist1.txt", "wordlist2.txt"},
        RulePaths:     []string{"rule1.rule", "rule2.rule"},
    }

    b.ResetTimer()
    for i := 0; i < b.N; i++ {
        _ = executor.buildHashcatCommand(assignment)
    }
}