rename functions to human-friendly

get_lan_ip: better algorithm

.archive/coro.py

@@ -2,13 +2,13 @@
 import typing
 import asyncio
 
-from findssh.coro import isportopen
+from findssh.coro import is_port_open
 
 
 async def as_completed(
     net: ip.IPv4Network, port: int, service: str, timeout: float
 ) -> typing.List[typing.Tuple[ip.IPv4Address, str]]:
-    futures = [isportopen(host, port, service) for host in net.hosts()]
+    futures = [is_port_open(host, port, service) for host in net.hosts()]
     hosts = []
     for future in asyncio.as_completed(futures, timeout=timeout):
         try:

.archive/findssh.m

@@ -17,7 +17,7 @@
 validateattributes(service, {'string', 'char'}, {'scalartext'})
 validateattributes(timeout, {'numeric'}, {'real', 'nonnegative'})
 
-net = py.findssh.netfromaddress(py.findssh.getLANip());
+net = py.findssh.address2net(py.findssh.get_lan_ip());
 
 hosts = cell(py.findssh.threadpool.get_hosts(net, uint8(port), service, timeout));
 

README.md

@@ -7,7 +7,10 @@
 Platform-independently find SSH servers (or other services with open ports) on an IPv4 subnet in pure Python WITHOUT NMAP.
 Scan entire IPv4 subnet in less than 1 second using Python standard library `asyncio`  coroutines and a single thread.
 
-The `asyncio` coroutine method uses ONE thread and is significantly *faster* than `concurrent.futures.ThreadPoolExecutor`, even (perhaps especially) with hundreds of threads in the ThreadPool.
+The default
+[asyncio coroutine](https://docs.python.org/3/library/asyncio.html)
+uses a single thread and is more than 10x faster than
+[concurrent.futures.ThreadPoolExecutor](https://docs.python.org/3/library/concurrent.futures.html).
 
 Although speed advantages weren't seen in our testing, `findssh` works with PyPy as well.
 
@@ -25,14 +28,16 @@ pip install -e findssh
 
 ## Usage
 
+A canonical way to use FindSSH from other Python scripts is [asyncio](example/asyncio_get_hosts.py).
+
+---
+
 from command line:
 
 ```sh
 python -m findssh
 ```
 
-### Command line options
-
 * `-s`  check the string from the server to attempt to verify the correct service has been found
 * `-t` timeout per server (seconds)  useful for high latency connection
 * `-b` baseip (check other subnet besides your own)

example/asyncio_get_hosts.py

@@ -0,0 +1,19 @@
+#!/usr/bin/env python3
+"""
+example of using findssh from a Python script
+"""
+
+import asyncio
+
+import findssh
+
+PORT = 22  # default SSH port
+TIMEOUT = 1.0  # seconds
+# timeout needs to be finite otherwise non-existant hosts are waited for forever
+
+ownIP = findssh.get_lan_ip()
+print("own address", ownIP)
+net = findssh.address2net(ownIP)
+print("searching", net)
+
+asyncio.run(findssh.get_hosts(net, PORT, TIMEOUT))

example/sequential_get_hosts.py

@@ -0,0 +1,21 @@
+#!/usr/bin/env python3
+"""
+example of using findssh from a Python script.
+
+Sequential non-parallel host discovery is impractically slow
+"""
+
+import findssh
+from findssh.base import get_hosts_seq
+
+PORT = 22  # default SSH port
+TIMEOUT = 1.0  # seconds
+# timeout needs to be finite otherwise non-existant hosts are waited for forever
+
+ownIP = findssh.get_lan_ip()
+print("own address", ownIP)
+net = findssh.address2net(ownIP)
+print("searching", net)
+
+for host in get_hosts_seq(net, PORT, TIMEOUT):
+    print(host)

example/thread_get_hosts.py

@@ -0,0 +1,21 @@
+#!/usr/bin/env python3
+"""
+example of using findssh from a Python script.
+
+Threads are MUCH slower than using asyncio as in asyncio_get_hosts.py
+"""
+
+import findssh
+import findssh.threadpool
+
+PORT = 22  # default SSH port
+TIMEOUT = 1.0  # seconds
+# timeout needs to be finite otherwise non-existant hosts are waited for forever
+
+ownIP = findssh.get_lan_ip()
+print("own address", ownIP)
+net = findssh.address2net(ownIP)
+print("searching", net)
+
+for host in findssh.threadpool.get_hosts(net, PORT, TIMEOUT):
+    print(host)

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "findssh"
-version = "1.6.0"
+version = "2.0.0"
 description = "find open servers on IPv4 subnet such as SSH"
 keywords = ["port-scan", "ssh"]
 classifiers = [

src/findssh/__init__.py

@@ -1,4 +1,30 @@
+from __future__ import annotations
+import socket
+import ipaddress
+
 from .coro import get_hosts
-from .base import netfromaddress, getLANip
 
-__all__ = ["get_hosts", "netfromaddress", "getLANip"]
+__all__ = ["get_hosts", "address2net", "get_lan_ip"]
+
+
+def get_lan_ip() -> ipaddress.IPv4Address | ipaddress.IPv6Address:
+    """
+    get IP address of currently used LAN interface
+    ref: http://stackoverflow.com/a/23822431
+    """
+
+    return ipaddress.ip_address(socket.gethostbyname(socket.gethostname()))
+
+
+def address2net(
+    addr: ipaddress.IPv4Address, mask: str = "24"
+) -> ipaddress.IPv4Network | ipaddress.IPv6Network:
+
+    if isinstance(addr, ipaddress.IPv4Address):
+        net = ipaddress.ip_network(addr.exploded.rsplit(".", 1)[0] + f".0/{mask}")
+    elif isinstance(addr, ipaddress.IPv6Address):
+        net = ipaddress.ip_network(addr.exploded.rsplit(":", 1)[0] + f":0/{mask}")
+    else:
+        raise TypeError(addr)
+
+    return net

src/findssh/__main__.py

@@ -17,7 +17,7 @@
 import ipaddress as ip
 from argparse import ArgumentParser
 
-from .base import getLANip, netfromaddress, get_hosts_seq
+from . import get_lan_ip, address2net
 from . import coro
 from . import threadpool
 
@@ -32,7 +32,11 @@ def main():
         "-s", "--service", default="", help="string to match to qualify detections"
     )
     p.add_argument(
-        "-t", "--timeout", help="timeout to wait for server", type=float, default=TIMEOUT
+        "-t",
+        "--timeout",
+        help="timeout to wait for server. Must be finite or will hang.",
+        type=float,
+        default=TIMEOUT,
     )
     p.add_argument("-b", "--baseip", help="set a specific subnet to scan")
     p.add_argument("-v", "--verbose", action="store_true")
@@ -45,21 +49,24 @@ def main():
         logging.basicConfig(level=logging.DEBUG)
 
     if not P.baseip:
-        ownip = getLANip()
+        ownip = get_lan_ip()
         print("own address", ownip)
     else:
         ownip = ip.ip_address(P.baseip)
 
-    net = netfromaddress(ownip)
+    net = address2net(ownip)
     print("searching", net)
 
     if P.threadpool:
-        for host in threadpool.get_hosts(net, P.port, P.service, P.timeout):
+        for host in threadpool.get_hosts(
+            net,
+            P.port,
+            P.timeout,
+            P.service,
+        ):
             print(host)
-    elif isinstance(net, ip.IPv6Network):
-        get_hosts_seq(net, P.port, P.service, P.timeout)
     else:
-        asyncio.run(coro.get_hosts(net, P.port, P.service, P.timeout))
+        asyncio.run(coro.get_hosts(net, P.port, P.timeout, P.service))
 
 
 if __name__ == "__main__":

src/findssh/base.py

@@ -4,55 +4,22 @@
 import typing as T
 
 
-def getLANip() -> ip.IPv4Address | ip.IPv6Address:
-    """get IP of own interface
-    ref: http://stackoverflow.com/a/23822431
-    """
-    with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as s:
-        s.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
-        # don't use link local here (169.254.x.x) unless you have a specific need
-        try:
-            s.connect(("<broadcast>", 0))
-        except OSError:
-            s.connect(("8.8.8.8", 80))  # for BSD/Mac
-
-        name = s.getsockname()[0]
-
-    return ip.ip_address(name)
-
-
-def validateservice(service: str, h: str, b: bytes) -> str:
+def get_service(b: bytes, service: str | None = None) -> str:
     """
     splitlines is in case the ASCII/UTF8 response is less than 32 bytes,
     hoping server sends a \r\n
     """
-    if not b:  # empty reply
-        return None
-    # %% non-empty reply
-    svc_txt = b.splitlines()[0].decode("utf-8", "ignore")
+
+    svc_txt = b.splitlines()[0].decode("utf-8", errors="ignore")
     # %% optional service validation
     if service and service not in svc_txt.lower():
         return None
 
     return svc_txt
 
 
-def netfromaddress(
-    addr: ip.IPv4Address, mask: str = "24"
-) -> ip.IPv4Network | ip.IPv6Network:
-
-    if isinstance(addr, ip.IPv4Address):
-        net = ip.ip_network(addr.exploded.rsplit(".", 1)[0] + f".0/{mask}")
-    elif isinstance(addr, ip.IPv6Address):
-        net = ip.ip_network(addr.exploded.rsplit(":", 1)[0] + f":0/{mask}")
-    else:
-        raise TypeError(addr)
-
-    return net
-
-
-def isportopen(
-    host: ip.IPv4Address, port: int, service: str, timeout: float
+def is_port_open(
+    host: ip.IPv4Address, port: int, timeout: float, service: str | None = None
 ) -> tuple[ip.IPv4Address, str]:
     """
     is a port open? Without coroutines.
@@ -67,23 +34,26 @@ def isportopen(
                 return None
         except socket.gaierror:
             return None
-        # %% service decode (optional)
+
         try:
-            if svc_txt := validateservice(service, h, s.recv(32)):
-                return host, svc_txt
+            if not (resp := s.recv(32)):
+                return None
         except (socket.timeout, ConnectionError):
             return None
 
+    if svc_txt := get_service(resp, service):
+        return host, svc_txt
+
     return None
 
 
 def get_hosts_seq(
-    net: ip.IPv4Network, port: int, service: str, timeout: float
+    net: ip.IPv4Network, port: int, timeout: float, service: str | None = None
 ) -> T.Iterable[tuple[ip.IPv4Address, str]]:
     """
     find hosts sequentially (no parallelism or concurrency)
     """
 
     for host in net.hosts():
-        if res := isportopen(host, port, service, timeout):
+        if res := is_port_open(host, port, timeout, service):
             yield res

src/findssh/coro.py

@@ -9,12 +9,20 @@
 import ipaddress as ip
 import asyncio
 
-from .base import validateservice
+from .base import get_service
+
+__all__ = ["get_hosts"]
 
 
 async def get_hosts(
-    net: ip.IPv4Network, port: int, service: str, timeout: float
+    net: ip.IPv4Network,
+    port: int,
+    timeout: float,
+    service: str | None = None,
 ) -> list[tuple[ip.IPv4Address, str]]:
+    """
+    Timeout must be finite otherwise non-existant hosts are waited for forever
+    """
 
     hosts = []
     for h in asyncio.as_completed(
@@ -31,13 +39,14 @@ async def waiter(
     host: ip.IPv4Address, port: int, service: str, timeout: float
 ) -> tuple[ip.IPv4Address, str]:
     try:
-        res = await asyncio.wait_for(isportopen(host, port, service), timeout=timeout)
+        res = await asyncio.wait_for(is_port_open(host, port, service), timeout=timeout)
     except asyncio.TimeoutError:
         res = None
+
     return res
 
 
-async def isportopen(
+async def is_port_open(
     host: ip.IPv4Address, port: int, service: str
 ) -> tuple[ip.IPv4Address, str]:
     """
@@ -47,11 +56,13 @@ async def isportopen(
 
     try:
         reader, _ = await asyncio.open_connection(host_str, port)
-        b = await reader.read(32)  # arbitrary number of bytes
+        if not (b := await reader.read(32)):
+            return None
     except OSError as err:  # to avoid flake8 error OSError has ConnectionError
         logging.debug(err)
         return None
-    # %% service decode (optional)
-    if svc_txt := validateservice(service, host_str, b):
+
+    if svc_txt := get_service(b, service):
         return host, svc_txt
+
     return None

src/findssh/tests/test_coro.py

@@ -2,15 +2,15 @@
 import asyncio
 
 import findssh
+import findssh.coro
 
 PORT = 22
-SERVICE = ""
-TIMEOUT = 1.0
+TIMEOUT = 0.5
 
 
 def test_coroutine():
-    net = findssh.netfromaddress(findssh.getLANip())
-    hosts = asyncio.run(findssh.get_hosts(net, PORT, SERVICE, TIMEOUT))
+    net = findssh.address2net(findssh.get_lan_ip())
+    hosts = asyncio.run(findssh.coro.get_hosts(net, PORT, TIMEOUT))
     for host, svc in hosts:
         assert isinstance(host, ipaddress.IPv4Address)
         assert isinstance(svc, str)