2601.01 - Adding stealth mode

portspoof_py/cli.py

@@ -43,6 +43,9 @@ def _parse_args(argv=None):
                    help='Skip iptables rule setup/teardown')
     p.add_argument('-v', '--verbose',   action='store_true',
                     help='Print each connection to stdout')
+    p.add_argument('--stealth',         action='store_true',
+                    help='Send no data on connections — close immediately, '
+                         'still logging each connection')
     p.add_argument('--admin-port',  type=int, default=None, metavar='PORT',
                    help='Enable web admin interface on this port (default: disabled)')
     p.add_argument('--admin-host',   default='127.0.0.1', metavar='HOST',
@@ -177,7 +180,8 @@ def main(argv=None) -> int:
         try:
             bind_ip: Optional[str] = args.bind_ip or None
             tasks.append(asyncio.create_task(
-                run_server(cfg, log, bind_ip, args.port, args.verbose, stop_event, settings)
+                run_server(cfg, log, bind_ip, args.port, args.verbose, stop_event, settings,
+                           stealth=args.stealth)
             ))
             if args.admin_port:
                 tasks.append(asyncio.create_task(

portspoof_py/server.py

@@ -38,6 +38,7 @@ async def _handle(
     log: Logger,
     verbose: bool,
     settings: Settings,
+    stealth: bool = False,
 ) -> None:
     sock = writer.get_extra_info('socket')
     peername = writer.get_extra_info('peername') or ('?.?.?.?', 0)
@@ -49,15 +50,19 @@ async def _handle(
         sockname = writer.get_extra_info('sockname') or ('', 0)
         original_port = sockname[1]
 
-    banner = cfg.get_banner(original_port)
+    banner = b'' if stealth else cfg.get_banner(original_port)
 
     if verbose:
+        if stealth:
+            print(f"[conn] {src_ip}:{src_port} → port {original_port}  banner=(none, stealth)")
+        else:
             print(
                 f"[conn] {src_ip}:{src_port} → port {original_port}"
                 f"  banner={banner[:32].hex()}{'…' if len(banner) > 32 else ''}"
             )
 
     try:
+        if not stealth:
             await asyncio.sleep(random.uniform(settings.delay_min, settings.delay_max))
             if banner:
                 writer.write(banner)
@@ -82,11 +87,12 @@ async def run_server(
     verbose: bool,
     stop_event: asyncio.Event,
     settings: Settings,
+    stealth: bool = False,
 ) -> None:
     """Start the asyncio TCP server and run until stop_event is set."""
 
     def handler(r, w):
-        asyncio.create_task(_handle(r, w, cfg, log, verbose, settings))
+        asyncio.create_task(_handle(r, w, cfg, log, verbose, settings, stealth))
 
     server = await asyncio.start_server(
         handler,
@@ -98,6 +104,8 @@ async def run_server(
 
     addrs = ', '.join(str(s.getsockname()) for s in server.sockets)
     print(f"[portspoof] listening on {addrs}")
+    if stealth:
+        print("[portspoof] stealth mode: connections closed without sending data")
 
     async with server:
         await stop_event.wait()