Minor improvements to Kerberos, IGMP and FWDM (#5019)

* kerberos: support password info in KPASSWD errors

AI-Assisted: no

* fwdm: support setting a remote port in server mode

AI-Assisted: no

* igmp: fix typo + implement IGMPv3 leave equivalent

AI-Assisted: no

Author: gpotter2

scapy/fwdmachine.py

@@ -74,6 +74,7 @@ class ForwardMachine:
     :param proto: the proto to use (default SOCK_STREAM)
     :param remote_address: the IP to use in SERVER mode, or by default in TPROXY when
         the destination is the local IP.
+    :param remote_port: the port to use in SERVER mode. (else use 'port')
     :param remote_af: (optional) if provided, use a different address family to connect
         to the remote host.
     :param bind_address: the IP to bind locally. "0.0.0.0" by default in SERVER mode,
@@ -108,6 +109,7 @@ def __init__(
         af: socket.AddressFamily = socket.AF_INET,
         proto: socket.SocketKind = socket.SOCK_STREAM,
         remote_address: str = None,
+        remote_port: int = None,
         remote_af: Optional[socket.AddressFamily] = None,
         bind_address: str = None,
         tls: bool = False,
@@ -129,6 +131,7 @@ def __init__(
         self.timeout = timeout
         self.MTU = MTU
         self.remote_address = remote_address
+        self.remote_port = remote_port
         if self.tls or self.af == 40:  # TLS or VSOCK
             self.sockcls = StreamSocketPeekless
         else:
@@ -164,9 +167,9 @@ def run(self):
             conn, addr = self.ssock.accept()
             # Calc dest
             dest = conn.getsockname()
-            if self.mode == ForwardMachine.MODE.SERVER or (
-                dest[0] in self.local_ips and self.remote_address
-            ):
+            if self.mode == ForwardMachine.MODE.SERVER:
+                dest = (self.remote_address, self.remote_port or self.port)
+            elif dest[0] in self.local_ips and self.remote_address:
                 dest = (self.remote_address,) + dest[1:]
             print(self.ct.green("%s -> %s connected !" % (repr(addr), repr(dest))))
             try:

scapy/layers/igmp.py

@@ -412,7 +412,7 @@ class IGMPv3_MRT(IGMPv3):
 
 
 bind_layers(IP, IGMP, proto=2)
-bind_top_down(IP, IGMP, proto=2, ttl=1, tox=0xC0)
+bind_top_down(IP, IGMP, proto=2, ttl=1, tos=0xC0)
 
 
 def _igmp_mq_addr(pkt):
@@ -455,14 +455,20 @@ def igmp_join(gaddr: str, version=2, psrc=None, iface=None):
 
 
 @conf.commands.register
-def igmp_leave(gaddr: str, psrc=None, iface=None):
+def igmp_leave(gaddr: str, version=2, psrc=None, iface=None):
     """
     Send a IGMP Leave Group to leave a multicast group
 
     :param gaddr: the IPv4 of the group to leave
     :param psrc: (optional) the source IP
     """
-    send(IP(src=psrc) / IGMPv2_LG(gaddr=gaddr), iface=iface)
+    if version == 1:
+        raise ValueError("IGMPv1 does not include a mechanism to leave !")
+    elif version == 2:
+        pkt = IP(src=psrc) / IGMPv2_LG(gaddr=gaddr)
+    elif version == 3:
+        pkt = IP(src=psrc) / IGMPv3_MR(records=[IGMPv3_MR_Group(rtype=3, maddr=gaddr)])
+    send(pkt, iface=iface)
 
 
 class IGMPMQResult(PacketList):

scapy/layers/kerberos.py

@@ -98,17 +98,19 @@
     FieldLenField,
     FlagsField,
     IntEnumField,
+    IntField,
     LEIntEnumField,
-    LenField,
     LEShortEnumField,
     LEShortField,
+    LenField,
     LongField,
     MayEnd,
     MultipleTypeField,
     PacketField,
     PacketLenField,
     PacketListField,
     PadField,
+    ScalingField,
     ShortEnumField,
     ShortField,
     StrField,
@@ -2848,10 +2850,55 @@ def answers(self, other):
 }
 
 
+class DOMAIN_PASSWORD_INFORMATION(Packet):
+    # [MS-SAMR] sect 2.2.3.5
+    fields_desc = [
+        IntField("MinPasswordLength", 0),
+        IntField("PasswordHistoryLength", 0),
+        FlagsField(
+            "PasswordProperties",
+            0,
+            32,
+            {
+                0x00000001: "DOMAIN_PASSWORD_COMPLEX",
+                0x00000002: "DOMAIN_PASSWORD_NO_ANON_CHANGE",
+                0x00000004: "DOMAIN_PASSWORD_NO_CLEAR_CHANGE",
+                0x00000008: "DOMAIN_LOCKOUT_ADMINS",
+                0x00000010: "DOMAIN_PASSWORD_STORE_CLEARTEXT",
+                0x00000020: "DOMAIN_REFUSE_PASSWORD_CHANGE",
+                0x00000040: "DOMAIN_NO_LM_OWF_CHANGE",
+            },
+        ),
+        ScalingField("MaxPasswordAge", 30 * 24 * 3600, scaling=1 / 1e7, fmt="!Q"),
+        ScalingField("MinPasswordAge", 0, scaling=1 / 1e7, fmt="!Q"),
+    ]
+
+
+class KPasswdResult(Packet):
+    # This is guessed from looking at MIT's implementation + ntsecapi.h
+    fields_desc = [
+        ShortField("PasswordInfoValid", 0),
+        PacketField(
+            "DomainPasswordInfo",
+            DOMAIN_PASSWORD_INFORMATION(),
+            DOMAIN_PASSWORD_INFORMATION,
+        ),
+    ]
+
+
+class _KPasswdRepDataResult_Field(StrField):
+    def m2i(self, pkt, s):
+        val = super(_KPasswdRepDataResult_Field, self).m2i(pkt, s)
+        if len(val or b"") == 30:
+            # A 30 octets blob is most likely the AD policy block
+            return KPasswdResult(val)
+        return val
+
+
 class KPasswdRepData(Packet):
     fields_desc = [
         ShortEnumField("resultCode", 0, KPASSWD_RESULTS),
-        StrField("resultString", ""),
+        _KPasswdRepDataResult_Field("resultString", ""),
     ]
 
 

test/scapy/layers/igmp.uts

@@ -11,6 +11,7 @@ b=IP(src="1.2.3.4")
 c=IGMP(gaddr="0.0.0.0")
 x = a/b/c
 assert x.mrcode == 20
+assert x[IP].tos == 0xc0
 assert x[IP].dst == "224.0.0.1"
 
 = Build IGMP - Custom membership
@@ -20,6 +21,7 @@ b=IP(src="1.2.3.4")
 c=IGMP(gaddr="224.0.1.2")
 x = a/b/c
 assert x.mrcode == 20
+assert x[IP].tos == 0xc0
 assert x[IP].dst == "224.0.1.2"
 
 = Build IGMP - LG
@@ -31,17 +33,20 @@ x = a/b/c
 x = Ether(bytes(x))
 assert x.dst == "01:00:5e:00:00:02"
 assert x.mrcode == 0
+assert x[IP].tos == 0xc0
 assert x[IP].dst == "224.0.0.2"
 
 = Change IGMP params
 
 x = Ether(src="00:01:02:03:04:05")/IP()/IGMP()
 assert x.mrcode == 20
+assert x[IP].tos == 0xc0
 assert x[IP].dst == "224.0.0.1"
 
 x = Ether(src="00:01:02:03:04:05")/IP()/IGMP(gaddr="224.2.3.4", type=0x12)
 x.mrcode = 1
 x = Ether(raw(x))
+assert x[IP].tos == 0xc0
 assert x.mrcode == 1
 
 x.gaddr = "224.3.2.4"