use Python >= 3.8 syntax

.flake8

@@ -1,5 +1,3 @@
 [flake8]
-max-line-length = 132
+max-line-length = 100
 exclude = .git,__pycache__,.eggs/,doc/,docs/,build/,dist/,.archive/
-per-file-ignores =
-    __init__.py:F401

pyproject.toml

@@ -35,7 +35,7 @@ lint = ["flake8", "flake8-bugbear", "flake8-builtins", "flake8-blind-except", "m
 readme = {file = ["README.md"], content-type = "text/markdown"}
 
 [tool.black]
-line-length = 100
+line-length = 90
 
 [tool.mypy]
 files = ["src"]

src/findssh/__init__.py

@@ -1,8 +1,4 @@
-import os
-import asyncio
-
 from .coro import get_hosts
 from .base import netfromaddress, getLANip
 
-if os.name == "nt":
-    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())  # type: ignore
+__all__ = ["get_hosts", "netfromaddress", "getLANip"]

src/findssh/__main__.py

@@ -18,7 +18,7 @@
 from argparse import ArgumentParser
 
 from .base import getLANip, netfromaddress, get_hosts_seq
-from .coro import get_hosts as coro_get_hosts
+from . import coro
 from . import threadpool
 
 PORT = 22
@@ -28,13 +28,17 @@
 def main():
     p = ArgumentParser("scan for hosts with open port, without NMAP")
     p.add_argument("-p", "--port", help="single port to try", default=PORT, type=int)
-    p.add_argument("-s", "--service", default="", help="string to match to qualify detections")
+    p.add_argument(
+        "-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
     )
     p.add_argument("-b", "--baseip", help="set a specific subnet to scan")
     p.add_argument("-v", "--verbose", action="store_true")
-    p.add_argument("-threadpool", help="use threadpool instead of asyncio", action="store_true")
+    p.add_argument(
+        "-threadpool", help="use threadpool instead of asyncio", action="store_true"
+    )
     P = p.parse_args()
 
     if P.verbose:
@@ -55,7 +59,7 @@ def main():
     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.service, P.timeout))
 
 
 if __name__ == "__main__":

src/findssh/base.py

@@ -37,7 +37,9 @@ def validateservice(service: str, h: str, b: bytes) -> str:
     return svc_txt
 
 
-def netfromaddress(addr: ip.IPv4Address, mask: str = "24") -> ip.IPv4Network | ip.IPv6Network:
+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}")
@@ -67,11 +69,11 @@ def isportopen(
             return None
         # %% service decode (optional)
         try:
-            svc_txt = validateservice(service, h, s.recv(32))
+            if svc_txt := validateservice(service, h, s.recv(32)):
+                return host, svc_txt
         except (socket.timeout, ConnectionError):
             return None
-    if svc_txt:
-        return host, svc_txt
+
     return None
 
 

src/findssh/coro.py

@@ -17,9 +17,10 @@ async def get_hosts(
 ) -> list[tuple[ip.IPv4Address, str]]:
 
     hosts = []
-    for h in asyncio.as_completed([waiter(host, port, service, timeout) for host in net.hosts()]):
-        host = await h
-        if host:
+    for h in asyncio.as_completed(
+        [waiter(host, port, service, timeout) for host in net.hosts()]
+    ):
+        if host := await h:
             print(host)
             hosts.append(host)
 
@@ -36,7 +37,9 @@ async def waiter(
     return res
 
 
-async def isportopen(host: ip.IPv4Address, port: int, service: str) -> tuple[ip.IPv4Address, str]:
+async def isportopen(
+    host: ip.IPv4Address, port: int, service: str
+) -> tuple[ip.IPv4Address, str]:
     """
     https://docs.python.org/3/library/asyncio-stream.html#asyncio.open_connection
     """
@@ -49,7 +52,6 @@ async def isportopen(host: ip.IPv4Address, port: int, service: str) -> tuple[ip.
         logging.debug(err)
         return None
     # %% service decode (optional)
-    svc_txt = validateservice(service, host_str, b)
-    if svc_txt:
+    if svc_txt := validateservice(service, host_str, b):
         return host, svc_txt
     return None

src/findssh/threadpool.py

@@ -21,7 +21,10 @@ def get_hosts(
 
     with concurrent.futures.ThreadPoolExecutor() as exc:
         try:
-            futures = (exc.submit(isportopen, host, port, service, timeout) for host in net.hosts())
+            futures = (
+                exc.submit(isportopen, host, port, service, timeout)
+                for host in net.hosts()
+            )
             for future in concurrent.futures.as_completed(futures):
                 if res := future.result():
                     yield res