Stabilize discovery workflow

2026-05-29 17:16:46 +08:00
parent 15772bc17e
commit 3981ffdf5e
40 changed files with 2039 additions and 211 deletions
@@ -36,18 +36,76 @@ class NetworkScanner
    /**
     * @return DiscoveredHost[]
     */
-    public function scan(NetworkRange $range): array
+    public function scan(NetworkRange $range, ?int $scanJobId = null): array
+    {
+        $result = $this->scanResumable(
+            $range,
+            $scanJobId ?? 0,
+            0,
+            static fn (): bool => false,
+            static function (array $state): void {
+            }
+        );
+
+        return $result['hosts'];
+    }
+
+    /**
+     * @return array{hosts: DiscoveredHost[], completed: bool, next_ip_index: int, total_ips: int, scanned_count: int, hosts_found: int}
+     */
+    public function scanResumable(
+        NetworkRange $range,
+        int $scanJobId,
+        int $startIndex,
+        callable $shouldStop,
+        callable $onProgress
+    ): array
    {
        $ips = $this->enumerateIps($range->cidr);
        $aliveHosts = [];
+        $scannedCount = 0;
+        $nextIndex = max(0, $startIndex);
+        $arpEntries = $this->arpTableReader->read();
+
+        $onProgress([
+            'next_ip_index' => $nextIndex,
+            'total_ips' => count($ips),
+            'scanned_count' => $scannedCount,
+            'hosts_found' => count($aliveHosts),
+        ]);
+
+        foreach ($ips as $index => $ip) {
+            if ($index < $nextIndex) {
+                continue;
+            }
+
+            if ($shouldStop()) {
+                return [
+                    'hosts' => $aliveHosts,
+                    'completed' => false,
+                    'next_ip_index' => $index,
+                    'total_ips' => count($ips),
+                    'scanned_count' => $scannedCount,
+                    'hosts_found' => count($aliveHosts),
+                ];
+            }
+
+            $nextIndex = $index + 1;
+            $scannedCount++;

-        foreach ($ips as $ip) {
            // Step 1: Ping sweep
            if (!$this->pingScanner->ping($ip)) {
+                $onProgress([
+                    'next_ip_index' => $nextIndex,
+                    'total_ips' => count($ips),
+                    'scanned_count' => $scannedCount,
+                    'hosts_found' => count($aliveHosts),
+                ]);
                continue;
            }

            $host = new DiscoveredHost();
+            $host->scanJobId = $scanJobId > 0 ? $scanJobId : null;
            $host->ipAddress = $ip;
            $host->firstSeen = new \DateTimeImmutable();
            $host->lastSeen = new \DateTimeImmutable();
@@ -66,20 +124,30 @@ class NetworkScanner
            // Step 5: Confidence score
            $host->confidence = $this->fingerprintService->calculateConfidence(count($openPorts), $hostname !== null);

-            $aliveHosts[] = $host;
-        }
-
-        // Step 6: ARP table enrichment
-        $arpEntries = $this->arpTableReader->read();
-        foreach ($aliveHosts as $host) {
            if (isset($arpEntries[$host->ipAddress])) {
                $host->macAddress = $arpEntries[$host->ipAddress]['mac'] ?? null;
                $host->vendor = $host->vendor ?: ($arpEntries[$host->ipAddress]['vendor'] ?? null);
            }
+
            $this->discoveredHostRepository->save($host);
+
+            $aliveHosts[] = $host;
+            $onProgress([
+                'next_ip_index' => $nextIndex,
+                'total_ips' => count($ips),
+                'scanned_count' => $scannedCount,
+                'hosts_found' => count($aliveHosts),
+            ]);
        }

-        return $aliveHosts;
+        return [
+            'hosts' => $aliveHosts,
+            'completed' => true,
+            'next_ip_index' => $nextIndex,
+            'total_ips' => count($ips),
+            'scanned_count' => $scannedCount,
+            'hosts_found' => count($aliveHosts),
+        ];
    }

    /**
@@ -87,31 +155,41 @@ class NetworkScanner
     */
    private function enumerateIps(string $cidr): array
    {
-        if (str_contains($cidr, '/32')) {
-            return [explode('/', $cidr)[0]];
+        if (!preg_match('#^(\d+\.\d+\.\d+\.\d+)/(\d{1,2})$#', $cidr, $matches)) {
+            throw new \InvalidArgumentException("Invalid IPv4 CIDR: {$cidr}");
        }

-        // Only handle /24 or larger for MVP (to avoid scanning huge ranges)
-        if (preg_match('#^(\d+\.\d+\.\d+)\.(\d+)/(\d+)$#', $cidr, $m)) {
-            $prefix = $m[1];
-            $suffix = (int)$m[2];
-            $mask = (int)$m[3];
-
-            if ($mask > 24) {
-                // Treat smaller than /24 as single IP
-                return [$cidr];
-            }
-
-            $offset = $mask === 24 ? 0 : $suffix;
-            $count = 2 ** (24 - $mask);
-            $ips = [];
-            for ($i = 1; $i <= $count; $i++) {
-                $ips[] = $prefix . '.' . ($offset + $i);
-            }
-            return $ips;
+        $baseIp = $matches[1];
+        $mask = (int)$matches[2];
+        if ($mask < 24) {
+            throw new \InvalidArgumentException('Network ranges larger than /24 are disabled for the MVP scanner');
+        }
+        if ($mask > 32) {
+            throw new \InvalidArgumentException("Invalid IPv4 CIDR mask: {$cidr}");
        }

-        return [];
+        $base = ip2long($baseIp);
+        if ($base === false) {
+            throw new \InvalidArgumentException("Invalid IPv4 address: {$baseIp}");
+        }
+
+        $base = (int)sprintf('%u', $base);
+        $hostCount = 2 ** (32 - $mask);
+        $network = $base & ((0xFFFFFFFF << (32 - $mask)) & 0xFFFFFFFF);
+
+        if ($mask === 32) {
+            return [long2ip($network)];
+        }
+
+        $start = $mask === 31 ? $network : $network + 1;
+        $end = $mask === 31 ? $network + 1 : $network + $hostCount - 2;
+
+        $ips = [];
+        for ($ip = $start; $ip <= $end; $ip++) {
+            $ips[] = long2ip($ip);
+        }
+
+        return $ips;
    }

    private function resolveHostname(string $ip): ?string
@@ -19,51 +19,19 @@ class TcpPortScanner
        $openPorts = [];
        $timeoutSec = $timeoutMs / 1000;

-        if (function_exists('curl_multi_init')) {
-            // Parallel scan with curl_multi
-            $chunks = array_chunk($ports, $maxConcurrency);
-            foreach ($chunks as $chunk) {
-                $multiHandle = curl_multi_init();
-                $handles = [];
-                foreach ($chunk as $port) {
-                    $ch = curl_init();
-                    curl_setopt($ch, CURLOPT_URL, "http://{$ip}:{$port}");
-                    curl_setopt($ch, CURLOPT_RETURNTRANSFER, true);
-                    curl_setopt($ch, CURLOPT_CONNECTTIMEOUT_MS, $timeoutMs);
-                    curl_setopt($ch, CURLOPT_TIMEOUT_MS, $timeoutMs);
-                    curl_setopt($ch, CURLOPT_NOBODY, true);
-                    curl_multi_add_handle($multiHandle, $ch);
-                    $handles[$port] = $ch;
-                }
-
-                // Execute
-                $running = null;
-                do {
-                    curl_multi_exec($multiHandle, $running);
-                    curl_multi_select($multiHandle, 0.1);
-                } while ($running > 0);
-
-                // Check results
-                foreach ($handles as $port => $ch) {
-                    $err = curl_errno($ch);
-                    // CURLE_COULDNT_CONNECT (7) = port closed, others might mean open or filtered
-                    if ($err !== CURLE_COULDNT_CONNECT) {
-                        $openPorts[] = $port;
-                    }
-                    curl_multi_remove_handle($multiHandle, $ch);
-                    curl_close($ch);
-                }
-
-                curl_multi_close($multiHandle);
-            }
-        } else {
-            // Sequential fallback
-            foreach ($ports as $port) {
-                $fp = @fsockopen($ip, $port, $errno, $errstr, $timeoutSec);
-                if ($fp !== false) {
-                    $openPorts[] = $port;
-                    fclose($fp);
-                }
+        foreach ($ports as $port) {
+            $errno = 0;
+            $errstr = '';
+            $fp = @stream_socket_client(
+                "tcp://{$ip}:{$port}",
+                $errno,
+                $errstr,
+                $timeoutSec,
+                STREAM_CLIENT_CONNECT
+            );
+            if ($fp !== false) {
+                $openPorts[] = $port;
+                fclose($fp);
            }
        }

@@ -6,14 +6,20 @@ namespace Domovoy\Services\Inventory;

 use Domovoy\Models\Device;
 use Domovoy\Repositories\DeviceRepository;
+use Domovoy\Repositories\DiscoveredHostRepository;

 class DeviceService
 {
    private DeviceRepository $deviceRepository;
+    private ?DiscoveredHostRepository $discoveredHostRepository;

-    public function __construct(DeviceRepository $deviceRepository)
+    public function __construct(
+        DeviceRepository $deviceRepository,
+        ?DiscoveredHostRepository $discoveredHostRepository = null
+    )
    {
        $this->deviceRepository = $deviceRepository;
+        $this->discoveredHostRepository = $discoveredHostRepository;
    }

    public function createFromDiscoveredHost(
@@ -56,4 +62,31 @@ class DeviceService
    {
        $this->deviceRepository->delete($id);
    }
+
+    public function mergeDiscoveredHost(int $hostId, int $deviceId): void
+    {
+        if ($this->discoveredHostRepository === null) {
+            throw new \RuntimeException('DiscoveredHostRepository is required for merge');
+        }
+
+        $host = $this->discoveredHostRepository->findById($hostId);
+        if ($host === null) {
+            throw new \RuntimeException('Discovered host not found');
+        }
+
+        $device = $this->deviceRepository->findById($deviceId);
+        if ($device === null) {
+            throw new \RuntimeException('Device not found');
+        }
+
+        $device->primaryIp = $device->primaryIp ?: $host->ipAddress;
+        $device->macAddress = $device->macAddress ?: $host->macAddress;
+        $device->hostname = $device->hostname ?: $host->hostname;
+        $device->vendor = $device->vendor ?: $host->vendor;
+        $this->deviceRepository->save($device);
+
+        $host->status = 'merged';
+        $host->matchedDeviceId = (string)$device->id;
+        $this->discoveredHostRepository->save($host);
+    }
 }
@@ -48,9 +48,15 @@ class MergeSuggestionService
            }
        }

-        // Low confidence: IP match
-        if ($host->primaryIp !== null) {
-            // Would need findByIp in repository — skip for now
+        if ($host->ipAddress !== '') {
+            $device = $this->deviceRepository->findByIp($host->ipAddress);
+            if ($device !== null) {
+                $suggestions[] = [
+                    'device' => $device,
+                    'confidence' => 40,
+                    'reason' => 'IP адрес совпадает',
+                ];
+            }
        }

        return $suggestions;
@@ -5,6 +5,7 @@ declare(strict_types=1);
 namespace Domovoy\Services\Jobs;

 use Domovoy\Models\ScanJob;
+use Domovoy\Repositories\NetworkRangeRepository;
 use Domovoy\Repositories\ScanJobRepository;
 use Domovoy\Services\Discovery\NetworkScanner;

@@ -12,13 +13,16 @@ class ScanJobRunner
 {
    private ScanJobRepository $scanJobRepository;
    private NetworkScanner $networkScanner;
+    private NetworkRangeRepository $networkRangeRepository;

    public function __construct(
        ScanJobRepository $scanJobRepository,
-        NetworkScanner $networkScanner
+        NetworkScanner $networkScanner,
+        NetworkRangeRepository $networkRangeRepository
    ) {
        $this->scanJobRepository = $scanJobRepository;
        $this->networkScanner = $networkScanner;
+        $this->networkRangeRepository = $networkRangeRepository;
    }

    /**
@@ -64,16 +68,25 @@ class ScanJobRunner
        $this->scanJobRepository->save($job);

        try {
+            $finalStatus = null;
            switch ($job->type) {
                case 'network_discovery':
-                    $this->runNetworkDiscovery($job);
+                    $finalStatus = $this->runNetworkDiscovery($job);
                    break;
                default:
                    throw new \RuntimeException("Unknown scan job type: {$job->type}");
            }

-            $job->status = 'done';
-            $job->finishedAt = new \DateTimeImmutable();
+            if ($finalStatus === 'done') {
+                $job->status = 'done';
+                $job->finishedAt = new \DateTimeImmutable();
+            } elseif ($finalStatus === 'paused') {
+                $job->status = 'paused';
+                $job->finishedAt = null;
+            } elseif ($finalStatus === 'cancelled') {
+                $job->status = 'cancelled';
+                $job->finishedAt = new \DateTimeImmutable();
+            }
        } catch (\Throwable $e) {
            $job->status = 'failed';
            $job->errorMessage = $e->getMessage();
@@ -83,13 +96,77 @@ class ScanJobRunner
        $this->scanJobRepository->save($job);
    }

-    private function runNetworkDiscovery(ScanJob $job): void
+    private function runNetworkDiscovery(ScanJob $job): string
    {
-        // For MVP, we scan all enabled network ranges if no specific range set
-        // TODO: filter by specific range when network_range_id is set
-        $settings = require dirname(__DIR__, 3) . '/../phinx.php';
+        if ($job->networkRangeId === null) {
+            throw new \RuntimeException('Network discovery job has no network_range_id');
+        }

-        // Use the ranges from database via NetworkRangeRepository
-        // For now, this is handled by the controller that creates the job with a specific range
+        $range = $this->networkRangeRepository->findById($job->networkRangeId);
+        if ($range === null) {
+            throw new \RuntimeException("Network range #{$job->networkRangeId} not found");
+        }
+        if (!$range->enabled) {
+            throw new \RuntimeException("Network range #{$job->networkRangeId} is disabled");
+        }
+
+        $existingProgress = $this->decodeProgress($job);
+        $startIndex = (int)($existingProgress['next_ip_index'] ?? 0);
+        $previousHostsFound = (int)($existingProgress['hosts_found'] ?? 0);
+
+        $result = $this->networkScanner->scanResumable(
+            $range,
+            $job->id ?? 0,
+            $startIndex,
+            function () use ($job): bool {
+                $fresh = $this->scanJobRepository->findById((int)$job->id);
+                return $fresh !== null && in_array($fresh->status, ['paused', 'cancelled'], true);
+            },
+            function (array $state) use ($job, $range, $previousHostsFound): void {
+                $fresh = $this->scanJobRepository->findById((int)$job->id);
+                if ($fresh !== null && in_array($fresh->status, ['paused', 'cancelled'], true)) {
+                    $job->status = $fresh->status;
+                }
+                $job->resultJson = json_encode([
+                    'network_range_id' => $range->id,
+                    'cidr' => $range->cidr,
+                    'next_ip_index' => $state['next_ip_index'],
+                    'total_ips' => $state['total_ips'],
+                    'scanned_count' => $state['scanned_count'],
+                    'hosts_found' => $previousHostsFound + $state['hosts_found'],
+                ], JSON_THROW_ON_ERROR);
+                $this->scanJobRepository->save($job);
+            }
+        );
+
+        $job->resultJson = json_encode([
+            'network_range_id' => $range->id,
+            'cidr' => $range->cidr,
+            'next_ip_index' => $result['next_ip_index'],
+            'total_ips' => $result['total_ips'],
+            'scanned_count' => $result['scanned_count'],
+            'hosts_found' => $previousHostsFound + $result['hosts_found'],
+        ], JSON_THROW_ON_ERROR);
+
+        if ($result['completed']) {
+            return 'done';
+        }
+
+        $fresh = $this->scanJobRepository->findById((int)$job->id);
+        if ($fresh !== null && in_array($fresh->status, ['paused', 'cancelled'], true)) {
+            return $fresh->status;
+        }
+
+        return 'paused';
+    }
+
+    private function decodeProgress(ScanJob $job): array
+    {
+        if ($job->resultJson === null || $job->resultJson === '') {
+            return [];
+        }
+
+        $decoded = json_decode($job->resultJson, true);
+        return is_array($decoded) ? $decoded : [];
    }
 }