Major refactoring of package structure

10 files changed, 522 insertions, 0 deletions

diff --git a/src/main/java/com/it_jaros/jscanner/scan/engine/CancelledToken.java b/src/main/java/com/it_jaros/jscanner/scan/engine/CancelledToken.java
new file mode 100644
index 0000000..1834a12
--- /dev/null
+++ b/src/main/java/com/it_jaros/jscanner/scan/engine/CancelledToken.java
@@ -0,0 +1,15 @@
+package com.it_jaros.jscanner.scan.engine;
+
+public class CancelledToken {
+    private volatile boolean cancelled = false;
+
+    public void cancel() {
+        if (!cancelled) {
+            this.cancelled = true;
+        }
+    }
+
+    public boolean isCancelled() {
+        return cancelled;
+    }
+}
diff --git a/src/main/java/com/it_jaros/jscanner/scan/engine/PollState.java b/src/main/java/com/it_jaros/jscanner/scan/engine/PollState.java
new file mode 100644
index 0000000..54ec767
--- /dev/null
+++ b/src/main/java/com/it_jaros/jscanner/scan/engine/PollState.java
@@ -0,0 +1,13 @@
+package com.it_jaros.jscanner.scan.engine;
+
+/**
+ * A sealed interface representing the three possible outcomes of a CompletionService
+ * poll operation: successful result, task failure, or not ready yet.
+ * Makes error handling explicit — Failure and Unavailable are distinct and compile-time
+ * required to handle via exhaustiveness checking in switch statements.
+ */
+public sealed interface PollState<T> {
+    record Success<T>(T value) implements PollState<T> {}
+    record Failure<T>(Throwable error) implements PollState<T> {}
+    record Unavailable<T>() implements PollState<T> {}
+}
diff --git a/src/main/java/com/it_jaros/jscanner/scan/engine/PortRangeIterator.java b/src/main/java/com/it_jaros/jscanner/scan/engine/PortRangeIterator.java
new file mode 100644
index 0000000..0673f0f
--- /dev/null
+++ b/src/main/java/com/it_jaros/jscanner/scan/engine/PortRangeIterator.java
@@ -0,0 +1,38 @@
+package com.it_jaros.jscanner.scan.engine;
+
+import java.util.BitSet;
+import java.util.Iterator;
+import java.util.NoSuchElementException;
+
+import static com.it_jaros.jscanner.scan.domain.PortRange.MAX_PORT;
+import static com.it_jaros.jscanner.scan.domain.PortRange.MIN_PORT;
+
+public class PortRangeIterator implements Iterator<Integer> {
+
+    private int currentPortCursor;
+    private int done = 0;
+    private final BitSet availablePorts = new BitSet(MAX_PORT);
+
+    public PortRangeIterator(BitSet specifiedPorts) {
+        availablePorts.or(specifiedPorts);
+        currentPortCursor = availablePorts.nextSetBit(MIN_PORT);
+    }
+
+    @Override
+    public Integer next() {
+        int p = availablePorts.nextSetBit(currentPortCursor);
+        if (p > 0) {
+            availablePorts.clear(p);
+            currentPortCursor = p + 1;
+            done++;
+            return p;
+        }
+
+        throw new NoSuchElementException("Reached end of port range");
+    }
+
+    @Override
+    public boolean hasNext() {
+        return availablePorts.nextSetBit(currentPortCursor) > 0;
+    }
+}
diff --git a/src/main/java/com/it_jaros/jscanner/scan/engine/PortResultAccumulator.java b/src/main/java/com/it_jaros/jscanner/scan/engine/PortResultAccumulator.java
new file mode 100644
index 0000000..0cc26b0
--- /dev/null
+++ b/src/main/java/com/it_jaros/jscanner/scan/engine/PortResultAccumulator.java
@@ -0,0 +1,49 @@
+package com.it_jaros.jscanner.scan.engine;
+
+import com.it_jaros.jscanner.scan.ExceptionInfo;
+import com.it_jaros.jscanner.scan.domain.*;
+import com.it_jaros.jscanner.scan.service.ServiceDetector;
+import com.it_jaros.jscanner.scan.service.ServiceType;
+
+import java.util.*;
+
+final class PortResultAccumulator {
+    private final String host;
+    private final BitSet openPorts = new BitSet(PortRange.MAX_PORT);
+    private final BitSet filteredPorts = new BitSet(PortRange.MAX_PORT);
+    private final Map<Integer, ServiceType> serviceTypes = new HashMap<>();
+    private final List<ScanFailure> scanFailures = new ArrayList<>();
+
+    PortResultAccumulator(String host) {
+        this.host = host;
+    }
+
+    void add(PortResult portResult) {
+        switch (portResult.getState()) {
+            case OPEN -> {
+                openPorts.set(portResult.getPort());
+                serviceTypes.put(portResult.getPort(), ServiceDetector.detect(portResult.getBanner()));
+            }
+            case FILTERED -> {
+                filteredPorts.set(portResult.getPort());
+            }
+            default -> {
+                // intentional no-op for uncovered port states
+            }
+        }
+        Exception e = portResult.getException();
+        if (e != null) {
+            scanFailures.add(new ScanFailure(portResult.getPort(), ExceptionInfo.from(e)));
+        }
+    }
+
+    ScanResult build() {
+        return new ScanResult(
+                host,
+                new PortList(openPorts),
+                new PortList(filteredPorts),
+                Collections.unmodifiableMap(serviceTypes),
+                Collections.unmodifiableList(scanFailures)
+        );
+    }
+}
diff --git a/src/main/java/com/it_jaros/jscanner/scan/engine/PortScanRateLimiter.java b/src/main/java/com/it_jaros/jscanner/scan/engine/PortScanRateLimiter.java
new file mode 100644
index 0000000..92b3bcb
--- /dev/null
+++ b/src/main/java/com/it_jaros/jscanner/scan/engine/PortScanRateLimiter.java
@@ -0,0 +1,37 @@
+package com.it_jaros.jscanner.scan.engine;
+
+import java.util.concurrent.locks.LockSupport;
+
+/**
+ * Per-host rate limiter. Each ScanHostTask creates its own instance and
+ * shares it with all its ScanPortTasks via constructor.
+ *
+ * Java allows one inner class to access another's private members, so this works.
+ */
+public class PortScanRateLimiter {
+    private final Object lock = new Object();
+    private volatile long nextAllowedTime;
+    private final long delayInNanos;
+    private final CancelledToken cancelledToken;
+
+    public PortScanRateLimiter(CancelledToken cancelledToken, long delayInNanos) {
+        this.cancelledToken = cancelledToken;
+        this.delayInNanos = delayInNanos;
+    }
+
+    void apply() { if (delayInNanos <= 0) {
+            return;
+        }
+        synchronized (lock) {
+            if (cancelledToken.isCancelled()) {
+                return;
+            }
+            long now = System.nanoTime();
+            if (nextAllowedTime > now) {
+                LockSupport.parkNanos(nextAllowedTime - now);
+                now = System.nanoTime();  // re-read after waking
+            }
+            nextAllowedTime = now + delayInNanos;
+        }
+    }
+}
diff --git a/src/main/java/com/it_jaros/jscanner/scan/engine/ProducerState.java b/src/main/java/com/it_jaros/jscanner/scan/engine/ProducerState.java
new file mode 100644
index 0000000..c49ea3e
--- /dev/null
+++ b/src/main/java/com/it_jaros/jscanner/scan/engine/ProducerState.java
@@ -0,0 +1,13 @@
+package com.it_jaros.jscanner.scan.engine;
+
+import java.util.concurrent.CompletionService;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public record ProducerState<OUTPUT>(
+        AtomicBoolean running,
+        AtomicInteger inPipeline,
+        Semaphore activeWorkers,
+        CompletionService<OUTPUT> completionService
+) {}
diff --git a/src/main/java/com/it_jaros/jscanner/scan/engine/ProducerThread.java b/src/main/java/com/it_jaros/jscanner/scan/engine/ProducerThread.java
new file mode 100644
index 0000000..e7bcbcc
--- /dev/null
+++ b/src/main/java/com/it_jaros/jscanner/scan/engine/ProducerThread.java
@@ -0,0 +1,83 @@
+package com.it_jaros.jscanner.scan.engine;
+
+import java.time.Duration;
+import java.util.Iterator;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Function;
+
+public class ProducerThread {
+
+    public static final Duration pollInterval = Duration.ofSeconds(1);
+
+    private final ExecutorService executor;
+    private final CancelledToken cancelledToken;
+
+    public ProducerThread(ScanExecutionContext context) {
+        this.executor = context.executorService();
+        this.cancelledToken = context.cancelledToken();
+    }
+
+    /**
+     * This method helps to cleanup the code a bit and remove redundancy
+     * The producer for providing hosts and the one for providing ports
+     * are similar and the small differences can be handled using a function
+     *
+     * @param queue
+     * @param maxWorkers
+     * @param taskFactory
+     * @param <INPUT>
+     * @param <OUTPUT>
+     * @return
+     */
+    public <INPUT, OUTPUT> ProducerState<OUTPUT> startProducer(
+            Iterator<INPUT> queue,
+            int maxWorkers,
+            Function<INPUT, Callable<OUTPUT>> taskFactory
+    ) {
+        final AtomicInteger inPipeline = new AtomicInteger(0);
+        final AtomicBoolean running = new AtomicBoolean(true);
+        final Semaphore activeWorkers = new Semaphore(maxWorkers);
+        CompletionService<OUTPUT> completionService = new ExecutorCompletionService<>(executor);
+        executor.submit(() -> {
+            try {
+                while (!cancelledToken.isCancelled() && queue.hasNext()) {
+                    // get semaphore and remember if task got submitted
+                    // so in case we fail to submit we release the semaphore
+                    activeWorkers.acquire();
+                    boolean isTaskSubmitted = false;
+                    try {
+                        // just in case something
+                        // changed while waiting
+                        if (cancelledToken.isCancelled()) {
+                            break;
+                        }
+
+                        // get next item and create callable
+                        // using lambda expression
+                        final INPUT item = queue.next();
+                        Callable<OUTPUT> task = taskFactory.apply(item);
+                        inPipeline.incrementAndGet();
+                        try {
+                            completionService.submit(task);
+                            isTaskSubmitted = true;
+                        } catch (Throwable e) {
+                            inPipeline.decrementAndGet();
+                            throw e;
+                        }
+                    } finally {
+                        if (!isTaskSubmitted) {
+                            activeWorkers.release();
+                        }
+                    }
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+            } finally {
+                running.set(false);
+            }
+        });
+        return new ProducerState<>(running, inPipeline, activeWorkers, completionService);
+    }
+}
diff --git a/src/main/java/com/it_jaros/jscanner/scan/engine/ScanExecutionContext.java b/src/main/java/com/it_jaros/jscanner/scan/engine/ScanExecutionContext.java
new file mode 100644
index 0000000..38e9cdb
--- /dev/null
+++ b/src/main/java/com/it_jaros/jscanner/scan/engine/ScanExecutionContext.java
@@ -0,0 +1,16 @@
+package com.it_jaros.jscanner.scan.engine;
+
+import com.it_jaros.jscanner.scan.Scan;
+import com.it_jaros.jscanner.scan.ScanOptions;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Semaphore;
+
+public record ScanExecutionContext(
+        ScanOptions scanOptions,
+        ExecutorService executorService,
+        Semaphore socketLimit,
+        CancelledToken cancelledToken,
+        Scan scan,
+        PortScanRateLimiter portScanRateLimiter
+) {}
diff --git a/src/main/java/com/it_jaros/jscanner/scan/engine/ScanHostTask.java b/src/main/java/com/it_jaros/jscanner/scan/engine/ScanHostTask.java
new file mode 100644
index 0000000..7a01d54
--- /dev/null
+++ b/src/main/java/com/it_jaros/jscanner/scan/engine/ScanHostTask.java
@@ -0,0 +1,123 @@
+package com.it_jaros.jscanner.scan.engine;
+
+import com.it_jaros.jscanner.scan.Scan;
+import com.it_jaros.jscanner.scan.domain.PortRange;
+import com.it_jaros.jscanner.scan.domain.PortResult;
+import com.it_jaros.jscanner.scan.domain.ScanResult;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.LockSupport;
+
+public class ScanHostTask implements Callable<ScanResult> {
+
+    private final Scan scan;
+    private final String host; // input parameter
+    private final ScanExecutionContext context;
+    private final CancelledToken cancelledToken;
+    private final boolean disableOnlineCheck;
+    private final int maxWorkersPerHost;
+    private final int timeoutInMillis;
+    private final long delayInNanos;
+
+    public ScanHostTask(Scan scan, String host, ScanExecutionContext context) {
+        this.scan = scan;
+        this.host = host;
+        this.context = context;
+        this.cancelledToken = context.cancelledToken();
+        this.delayInNanos = TimeUnit.MILLISECONDS.toNanos(Math.max(0, context.scanOptions().delayInMillis()));
+        this.disableOnlineCheck = context.scanOptions().disableOnlineCheck();
+        this.maxWorkersPerHost = context.scanOptions().maxWorkersPerHost();
+        this.timeoutInMillis = context.scanOptions().timeoutInMillis();
+    }
+
+    @Override
+    public ScanResult call() {
+        try {
+            scan.hostStart();
+            if (context.cancelledToken().isCancelled()) {
+                return ScanResult.empty(host);
+            }
+            return scanHostPorts();
+        } finally {
+            scan.hostFinish();
+        }
+    }
+
+    private ScanResult scanHostPorts() {
+        if (!disableOnlineCheck) {
+            boolean isHostOnline = checkHostOnline();
+            if (!isHostOnline) {
+                // Unreachable host
+                return ScanResult.empty(host);
+            }
+            // online check also sends packets to the target system.
+            // in order not to violate set delay time
+            // we wait here too
+            LockSupport.parkNanos(delayInNanos);
+        }
+
+        final PortRange portRange = new PortRange(scan.getPorts());
+        // producer thread
+        scan.producerStart();
+        ProducerState<PortResult> state = new ProducerThread(context).startProducer(
+                portRange.iterator(),
+                maxWorkersPerHost,
+                port -> new ScanPortTask(host, port, context)
+        );
+
+        // consumer is the main thread
+        // we run as long as the producer is running OR
+        // as long as things are in pipeline waiting to be processed
+        // ONLY exception is when cancelled is set
+        final PortResultAccumulator accumulator = new PortResultAccumulator(host);
+        while (!cancelledToken.isCancelled() && (state.running().get() || state.inPipeline().get() > 0)) {
+            try {
+                PollState<PortResult> poll = getPortResult(state);
+                if (poll instanceof PollState.Success<PortResult>(PortResult value)) {
+                    accumulator.add(value);
+                } else if (poll instanceof PollState.Failure(Throwable error)) {
+                    System.err.printf("scanHostPorts(%s): ScanPortTask() failed for with error %s -> %s%n", host, error.getClass().getSimpleName(), error.getMessage());
+                }
+            } catch (InterruptedException ignored) {
+                Thread.currentThread().interrupt();
+            }
+        }
+        scan.producerStop();
+
+        return accumulator.build();
+    }
+
+    private boolean checkHostOnline() {
+        try {
+            return InetAddress.getByName(host).isReachable(timeoutInMillis);
+        } catch (IOException e) {
+            // we ignore this error because it means that the host is probably not online
+        }
+
+        return false;
+    }
+
+    private PollState<PortResult> getPortResult(ProducerState<PortResult> state) throws InterruptedException {
+        Future<PortResult> portResultFuture = state.completionService().poll(ProducerThread.pollInterval.toMillis(), TimeUnit.MILLISECONDS);
+        if (portResultFuture == null) {
+            return new PollState.Unavailable<>();
+        }
+
+        PortResult portResult;
+        try {
+            portResult = portResultFuture.get();
+            return new PollState.Success<>(portResult);
+        } catch (ExecutionException e) {
+            Throwable cause = e.getCause();
+            return new PollState.Failure<>(cause);
+        } finally {
+            state.activeWorkers().release();
+            state.inPipeline().decrementAndGet();
+        }
+    }
+}
diff --git a/src/main/java/com/it_jaros/jscanner/scan/engine/ScanPortTask.java b/src/main/java/com/it_jaros/jscanner/scan/engine/ScanPortTask.java
new file mode 100644
index 0000000..873ec95
--- /dev/null
+++ b/src/main/java/com/it_jaros/jscanner/scan/engine/ScanPortTask.java
@@ -0,0 +1,135 @@
+package com.it_jaros.jscanner.scan.engine;
+
+import com.it_jaros.jscanner.scan.Scan;
+import com.it_jaros.jscanner.scan.domain.PortResult;
+import com.it_jaros.jscanner.scan.domain.PortState;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+import java.net.ConnectException;
+import java.net.InetSocketAddress;
+import java.net.NoRouteToHostException;
+import java.nio.ByteBuffer;
+import java.nio.channels.SocketChannel;
+import java.util.concurrent.Callable;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.locks.LockSupport;
+
+public class ScanPortTask implements Callable<PortResult> {
+
+    private static final int READ_BUFFER_SIZE = 1024;
+
+    private final CancelledToken cancelledToken;
+    private final PortScanRateLimiter portScanRateLimiter;  // per-host shared limiter
+    private final Scan scan;
+    private final Semaphore socketLimit;
+    private final String host;
+    private final boolean bannerRecognition;
+    private final int port;
+    private final long timeoutInNanos;
+
+    ScanPortTask(String host, int port, ScanExecutionContext context) {
+        this.host = host;
+        this.port = port;
+        this.bannerRecognition = context.scanOptions().bannerRecognition();
+        this.cancelledToken = context.cancelledToken();
+        this.portScanRateLimiter = context.portScanRateLimiter();
+        this.scan = context.scan();
+        this.socketLimit = context.socketLimit();
+        this.timeoutInNanos = TimeUnit.MILLISECONDS.toNanos(context.scanOptions().timeoutInMillis());
+    }
+
+    @Override
+    public PortResult call() throws Exception {
+        try {
+            socketLimit.acquire();
+            scan.portStart();
+            portScanRateLimiter.apply();
+            if (cancelledToken.isCancelled()) {
+                return PortResult.empty();
+            }
+            return checkPort();
+        } finally {
+            scan.portFinish();
+            socketLimit.release();
+        }
+    }
+
+    private PortResult checkPort() {
+        PortResult result = new PortResult();
+        result.setPort(port);
+        result.setState(PortState.UNKNOWN);
+        try(SocketChannel socketChannel = SocketChannel.open()) {
+            socketChannel.configureBlocking(false);
+            socketChannel.connect(new InetSocketAddress(host, port));
+            final long deadlineNanos = System.nanoTime() + timeoutInNanos;
+            final long waitInNanos = TimeUnit.MILLISECONDS.toNanos(1000);
+            boolean isConnected = socketChannel.finishConnect();
+            while (!cancelledToken.isCancelled() && !isConnected) {
+                long remainingNanos = deadlineNanos - System.nanoTime();
+                if (remainingNanos <= 0) {
+                    break;
+                }
+                LockSupport.parkNanos(Math.min(waitInNanos, remainingNanos));
+                isConnected = socketChannel.finishConnect();
+            }
+
+            if (cancelledToken.isCancelled()) {
+                return result;
+            }
+
+            if (isConnected) {
+                result.setState(PortState.OPEN);
+                if (bannerRecognition) {
+                    result.setBanner(getBanner(socketChannel));
+                }
+            } else {
+                result.setState(PortState.FILTERED);
+            }
+        } catch (ConnectException ignored) {
+            result.setState(PortState.CLOSED);
+        } catch (NoRouteToHostException ignored) {
+            // NoRouteToHostException: this can be safely ignored because the port is closed if a host is unreachable
+            // Will happen a lot when scanning for open ports, so not needed
+        } catch (IOException e) {
+            result.setException(e);
+        }
+
+        return result;
+    }
+
+    private byte[] getBanner(SocketChannel socketChannel) {
+        try {
+            return tryReadFrom(socketChannel);
+        } catch (IOException ignore) {
+            // ignore
+        }
+
+        return null;
+    }
+
+    private byte[] tryReadFrom(SocketChannel socketChannel) throws IOException {
+        byte[] buffer = new byte[READ_BUFFER_SIZE];
+        ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream();
+
+        boolean timedout = false;
+        long deadlineNanos = System.nanoTime() + timeoutInNanos;
+        while (!timedout) {
+            int bytesRead = socketChannel.read(ByteBuffer.wrap(buffer));
+            if (bytesRead == -1) {
+                break;
+            }
+            if (bytesRead > 0) {
+                byteArrayOutputStream.write(buffer, 0, bytesRead);
+            }
+            long remaining = deadlineNanos - System.nanoTime();
+            if (remaining <= 0) {
+                timedout = true;
+            } else {
+                LockSupport.parkNanos(Math.min(remaining, TimeUnit.MILLISECONDS.toNanos(1000)));
+            }
+        }
+        return byteArrayOutputStream.toByteArray();
+    }
+}