Add nbns_request() util

scapy/layers/dns.py

@@ -1405,7 +1405,7 @@ def dns_resolve(qname, qtype="A", raw=False, tcp=False, verbose=1, timeout=3, **
     :param timeout: seconds until timeout (per server)
     :raise TimeoutError: if no DNS servers were reached in time.
     """
-    # Unify types
+    # Unify types (for caching)
     qtype = DNSQR.qtype.any2i_one(None, qtype)
     qname = DNSQR.qname.any2i(None, qname)
     # Check cache

scapy/layers/netbios.py

@@ -34,9 +34,17 @@
     XShortField,
     XStrFixedLenField
 )
+from scapy.interfaces import _GlobInterfaceType
+from scapy.sendrecv import sr1
 from scapy.layers.inet import IP, UDP, TCP
 from scapy.layers.l2 import Ether, SourceMACField
 
+# Typing imports
+from typing import (
+    List,
+    Union,
+)
+
 
 class NetBIOS_DS(Packet):
     name = "NetBIOS datagram service"
@@ -398,6 +406,93 @@ def tcp_reassemble(cls, data, *args, **kwargs):
 bind_layers(TCP, NBTSession, dport=139, sport=139)
 
 
+_nbns_cache = conf.netcache.new_cache("nbns_cache", 300)
+
+
+@conf.commands.register
+def nbns_resolve(
+    qname: str,
+    iface: Union[_GlobInterfaceType, List[_GlobInterfaceType]] = None,
+    raw: bool = False,
+    timeout: int = 3,
+    **kwargs,
+) -> List[str]:
+    """
+    Perform a simple NBNS (NetBios Name Services) resolution with caching
+
+    :param qname: the name to query
+    :param iface: the interfaces to use. (default: all)
+    :param raw: return the whole netbios packet (default False)
+    :param timeout: seconds until timeout (per server)
+    :raise TimeoutError: if no DNS servers were reached in time.
+    """
+    kwargs.setdefault("verbose", 0)
+
+    # Unify types (for caching)
+    qname = NBNSQueryRequest.QUESTION_NAME.any2i(None, qname)
+
+    # Check cache
+    cache_ident = qname + b"raw" if raw else b""
+    result = _nbns_cache.get(cache_ident)
+    if result:
+        return result
+
+    if iface is None:
+        ifaces = [
+            x
+            for name, x in conf.ifaces.items()
+            if x.is_valid() and name != conf.loopback_name
+        ]
+    elif isinstance(iface, list):
+        ifaces = iface
+    else:
+        ifaces = [iface]
+
+    # Builds a request for each broadcast address of each interface
+    requests = []
+    for iface in ifaces:
+        for bdcst in conf.route.get_if_bcast(iface):
+            if bdcst == "255.255.255.255":
+                continue
+            requests.append(
+                IP(dst=bdcst) /
+                UDP() /
+                NBNSHeader() /
+                NBNSQueryRequest(QUESTION_NAME=qname)
+            )
+
+    if not requests:
+        return None
+
+    # Perform requests, get the first response
+    try:
+        old_checkIPAddr = conf.checkIPaddr
+        conf.checkIPaddr = False
+
+        res = sr1(
+            requests,
+            timeout=timeout,
+            first=True,
+            **kwargs,
+        )
+    finally:
+        conf.checkIPaddr = old_checkIPAddr
+
+    if res is not None:
+        if raw:
+            # Raw
+            result = res
+        else:
+            # Get IP
+            result = [x.NB_ADDRESS for x in res.ADDR_ENTRY]
+        if result:
+            # Cache it
+            _nbns_cache[cache_ident] = result
+        return result
+    else:
+        raise TimeoutError
+
+
 class NBNS_am(AnsweringMachine):
     function_name = "nbnsd"
     filter = "udp port 137"

scapy/route.py

@@ -229,8 +229,11 @@ def route(self, dst=None, dev=None, verbose=conf.verb, _internal=False):
 
     def get_if_bcast(self, iff):
         # type: (str) -> List[str]
+        """
+        Return the list of broadcast addresses of an interface.
+        """
         bcast_list = []
-        for net, msk, gw, iface, addr, metric in self.routes:
+        for net, msk, _, iface, _, _ in self.routes:
             if net == 0:
                 continue    # Ignore default route "0.0.0.0"
             elif msk == 0xffffffff:

scapy/sendrecv.py

@@ -86,6 +86,7 @@ class debug:
         if negative, how many times to retry when no more packets
         are answered
     :param multi: whether to accept multiple answers for the same stimulus
+    :param first: stop after receiving the first response of any sent packet
     :param rcv_pks: if set, will be used instead of pks to receive packets.
         packets will still be sent through pks
     :param prebuild: pre-build the packets before starting to send them.
@@ -125,6 +126,7 @@ def __init__(self,
                  chainCC=False,  # type: bool
                  retry=0,  # type: int
                  multi=False,  # type: bool
+                 first=False,  # type: bool
                  rcv_pks=None,  # type: Optional[SuperSocket]
                  prebuild=False,  # type: bool
                  _flood=None,  # type: Optional[_FloodGenerator]
@@ -150,6 +152,7 @@ def __init__(self,
         self.chainCC = chainCC
         self.multi = multi
         self.timeout = timeout
+        self.first = first
         self.session = session
         self.chainEX = chainEX
         self.stop_filter = stop_filter
@@ -254,7 +257,10 @@ def results(self):
 
     def _stop_sniffer_if_done(self) -> None:
         """Close the sniffer if all expected answers have been received"""
-        if self._send_done and self.noans >= self.notans and not self.multi:
+        if (
+            self._send_done and self.noans >= self.notans and not self.multi or
+            self.first and self.noans
+        ):
             if self.sniffer and self.sniffer.running:
                 self.sniffer.stop(join=False)