Merge pull request #3 from Nisitay/feature/fingerprint_uptime

Feature/fingerprint uptime

Author: Nisitay

CHANGELOG.md

@@ -1,5 +1,9 @@
 # Release History
 
+## pyp0f 0.3.0
+
+* Added TCP timestamps uptime detection.
+
 ## pyp0f 0.2.1
 
 * Added impersonation utility.

README.md

@@ -26,10 +26,10 @@ $ pip install pyp0f
 ## Features
 - Full p0f fingerprinting (MTU, TCP, HTTP)
 - p0f spoofing - impersonation (MTU, TCP)
+- TCP timestamps uptime detection
 
 ## In Progress
 - Flow tracking
-- TCP uptime detection
 - NAT detection
 
 ## Getting Started

docs/README.md

@@ -132,6 +132,39 @@ print(result.match)
 
 </details>
 
+</details>
+
+<details markdown="1">
+<summary>TCP uptime fingerprinting example</summary>
+
+```python
+from scapy.layers.inet import IP, TCP
+from pyp0f.net.packet import parse_packet
+from pyp0f.net.signatures import TCPPacketSignature
+from pyp0f.fingerprint.uptime import fingerprint_uptime
+from pyp0f.utils.time import get_unix_time_ms
+
+last_timestamp = IP() / TCP(seq=1, options=[("Timestamp", (1545573, 0))])
+current_timestamp = IP() / TCP(seq=2, options=[("Timestamp", (1545586, 0))])
+
+last_packet_signature = TCPPacketSignature.from_packet(parse_packet(last_timestamp))
+
+# Simulate different receive time
+last_packet_signature.received = get_unix_time_ms() - 130
+
+result = fingerprint_uptime(current_timestamp, last_packet_signature)
+print(result.tps)  # 100
+print(result.uptime)
+# Uptime(
+#     raw_frequency=107.6923076923077,
+#     frequency=100,
+#     total_minutes=257,
+#     modulo_days=497
+# )
+```
+
+</details>
+
 ## Impersonation
 `pyp0f` provides functionality to modify Scapy packets so that `p0f` will think it has been sent by a specific OS.
 
@@ -175,7 +208,7 @@ result = fingerprint_tcp(impersonated_packet)  # TCPResult for "s:unix:OpenVMS:7
 </details>
 
 ## Real World Examples
-`pyp0f` can be used in real world scenarios, whether its to passivly fingerprint remote hosts,
+`pyp0f` can be used in real world scenarios, whether its to passively fingerprint remote hosts,
 or to deceive remote `p0f`.
 
 ### Sniff connection attempts

pyp0f/__init__.py

@@ -1 +1 @@
-VERSION = "0.2.1"
+VERSION = "0.3.0"

pyp0f/database/records_database.py

@@ -86,7 +86,7 @@ def iter_values(
         self, key: Type[T], direction: Optional[Direction] = None
     ) -> Iterator[T]:
         """
-        Safely iterate a list of values.
+        Iterate a list of values.
         """
         try:
             values = self._get(key, direction)

pyp0f/fingerprint/__init__.py

@@ -1,5 +1,11 @@
-from .http import fingerprint as fingerprint_http
-from .mtu import fingerprint as fingerprint_mtu
-from .tcp import fingerprint as fingerprint_tcp
+from .http import fingerprint_http
+from .mtu import fingerprint_mtu
+from .tcp import fingerprint_tcp
+from .uptime import fingerprint_uptime
 
-__all__ = ["fingerprint_mtu", "fingerprint_tcp", "fingerprint_http"]
+__all__ = [
+    "fingerprint_mtu",
+    "fingerprint_tcp",
+    "fingerprint_http",
+    "fingerprint_uptime",
+]

pyp0f/fingerprint/http.py

@@ -20,7 +20,7 @@ def headers_match(
     i = 0  # Index of packet header
 
     for header in signature_headers:
-        orig_index = i
+        original_index = i
 
         while (
             i < len(packet_headers)
@@ -34,12 +34,12 @@ def headers_match(
 
             # Optional header -> check that it doesn't appear anywhere else
             if any(
-                header.lower_name == pkt_header.lower_name
-                for pkt_header in packet_headers
+                header.lower_name == packet_header.lower_name
+                for packet_header in packet_headers
             ):
                 return False
 
-            i = orig_index
+            i = original_index
             continue
 
         # Header found, validate values
@@ -49,15 +49,15 @@ def headers_match(
     return True
 
 
-def signatures_match(
+def http_signatures_match(
     signature: HTTPSignature, packet_signature: HTTPPacketSignature
 ) -> bool:
     """
     Check if HTTP signatures match by comparing the following criterias:
         - HTTP versions match.
         - All non-optional signature headers appear in the packet.
         - Absent headers in signature don't appear in the packet.
-        - Order and values of headers match (this is relatively slow).
+        - Order and values of headers match (relatively slow, hence why its last step).
     """
     packet_headers = packet_signature.header_names
     return (
@@ -68,7 +68,7 @@ def signatures_match(
     )
 
 
-def find_match(
+def find_http_match(
     packet_signature: HTTPPacketSignature,
     direction: Direction,
     database: Database,
@@ -79,7 +79,7 @@ def find_match(
     generic_match: Optional[HTTPRecord] = None
 
     for http_record in database.iter_values(HTTPRecord, direction):
-        if not signatures_match(http_record.signature, packet_signature):
+        if not http_signatures_match(http_record.signature, packet_signature):
             continue
 
         if not http_record.is_generic:
@@ -91,7 +91,7 @@ def find_match(
     return generic_match
 
 
-def fingerprint(buffer: BufferLike, options: Options = OPTIONS) -> HTTPResult:
+def fingerprint_http(buffer: BufferLike, *, options: Options = OPTIONS) -> HTTPResult:
     """
     Fingerprint the given HTTP 1.x payload.
 
@@ -111,5 +111,5 @@ def fingerprint(buffer: BufferLike, options: Options = OPTIONS) -> HTTPResult:
     return HTTPResult(
         buffer,
         packet_signature,
-        find_match(packet_signature, direction, options.database),
+        find_http_match(packet_signature, direction, options.database),
     )

pyp0f/fingerprint/mtu.py

@@ -5,20 +5,29 @@
 from pyp0f.database.signatures import MTUSignature
 from pyp0f.exceptions import PacketError
 from pyp0f.fingerprint.results import MTUResult
+from pyp0f.net.layers.tcp import TCPFlag
 from pyp0f.net.packet import Packet, PacketLike, parse_packet
 from pyp0f.net.signatures import MTUPacketSignature
 from pyp0f.options import OPTIONS, Options
 
 
-def valid_for_fingerprint(packet: Packet) -> bool:
+def valid_for_mtu_fingerprint(packet: Packet) -> bool:
     """
     Check if the given packet is valid for MTU fingerprint.
-    Packets with MSS value are valid for fingerprint.
+    SYN/SYN+ACK packets with MSS value are valid for fingerprint.
     """
-    return packet.should_fingerprint and packet.tcp.options.mss > 0
+    return (
+        packet.should_fingerprint
+        and packet.tcp.options.mss > 0
+        and packet.tcp.type
+        in (
+            TCPFlag.SYN,
+            TCPFlag.SYN | TCPFlag.ACK,
+        )
+    )
 
 
-def signatures_match(
+def mtu_signatures_match(
     signature: MTUSignature, packet_signature: MTUPacketSignature
 ) -> bool:
     """
@@ -27,23 +36,19 @@ def signatures_match(
     return signature.mtu == packet_signature.mtu
 
 
-def find_match(
+def find_mtu_match(
     packet_signature: MTUPacketSignature, database: Database
 ) -> Optional[MTURecord]:
     """
     Search through the database for a match for the given MTU signature.
     """
-    return next(
-        (
-            mtu_record
-            for mtu_record in database.iter_values(MTURecord)
-            if signatures_match(mtu_record.signature, packet_signature)
-        ),
-        None,
-    )
+    for mtu_record in database.iter_values(MTURecord):
+        if mtu_signatures_match(mtu_record.signature, packet_signature):
+            return mtu_record
+    return None
 
 
-def fingerprint(packet: PacketLike, options: Options = OPTIONS) -> MTUResult:
+def fingerprint_mtu(packet: PacketLike, *, options: Options = OPTIONS) -> MTUResult:
     """
     Fingerprint the given packet for MTU.
 
@@ -57,13 +62,16 @@ def fingerprint(packet: PacketLike, options: Options = OPTIONS) -> MTUResult:
     Returns:
         MTU fingerprint result
     """
-    parsed_packet = parse_packet(packet)
+    packet = parse_packet(packet)
 
-    if not valid_for_fingerprint(parsed_packet):
-        raise PacketError("Packet is invalid for MTU fingerprint")
+    if not valid_for_mtu_fingerprint(packet):
+        raise PacketError(
+            "Packet is invalid for MTU fingerprint. "
+            "Packet must be SYN/SYN+ACK with MSS value."
+        )
 
-    packet_signature = MTUPacketSignature.from_packet(parsed_packet)
+    packet_signature = MTUPacketSignature.from_packet(packet)
 
     return MTUResult(
-        parsed_packet, packet_signature, find_match(packet_signature, options.database)
+        packet, packet_signature, find_mtu_match(packet_signature, options.database)
     )

pyp0f/fingerprint/results/__init__.py

@@ -2,5 +2,16 @@
 from .http import HTTPResult
 from .mtu import MTUResult
 from .tcp import TCPMatch, TCPMatchType, TCPResult
+from .uptime import BAD_TPS, Uptime, UptimeResult
 
-__all__ = ["Result", "MTUResult", "TCPResult", "HTTPResult", "TCPMatchType", "TCPMatch"]
+__all__ = [
+    "Result",
+    "MTUResult",
+    "TCPResult",
+    "HTTPResult",
+    "TCPMatchType",
+    "TCPMatch",
+    "Uptime",
+    "UptimeResult",
+    "BAD_TPS",
+]

pyp0f/fingerprint/results/base.py

@@ -14,14 +14,14 @@
 @dataclass
 class Result(Generic[TMatch, TSignature], metaclass=ABCMeta):
     """
-    Fingerprint result
+    Fingerprint result.
     """
 
     packet: Packet
-    """Origin packet"""
+    """Origin packet."""
 
     packet_signature: TSignature
-    """Origin packet signature"""
+    """Origin packet signature."""
 
     match: Optional[TMatch] = None
-    """Fingerprint match, if any"""
+    """Fingerprint match, if any."""

pyp0f/fingerprint/results/http.py

@@ -12,11 +12,14 @@
 @dataclass
 class HTTPResult(Result[HTTPRecord, HTTPPacketSignature]):
     """
-    HTTP fingerprint result
+    HTTP fingerprint result.
     """
 
     packet: BufferLike
+    """Origin HTTP payload."""
+
     dishonest: bool = field(init=False)
+    """Software string (User-Agent or Server) looks forged?"""
 
     def __post_init__(self):
         self.dishonest = (

pyp0f/fingerprint/results/mtu.py

@@ -11,5 +11,5 @@
 @dataclass
 class MTUResult(Result[MTURecord, MTUPacketSignature]):
     """
-    MTU fingerprint result
+    MTU fingerprint result.
     """

pyp0f/fingerprint/results/tcp.py

@@ -18,29 +18,30 @@ class TCPMatchType(Enum):
 @dataclass
 class TCPMatch:
     """
-    Match for a TCP fingerprint
+    Match for a TCP fingerprint.
     """
 
     type: TCPMatchType
-    """Match type"""
+    """Match type."""
 
     record: TCPRecord
-    """Matched record"""
+    """Matched record."""
 
     @property
     def is_fuzzy(self) -> bool:
+        """Approximate match?"""
         return self.type != TCPMatchType.EXACT
 
 
 @add_slots
 @dataclass
 class TCPResult(Result[TCPMatch, TCPPacketSignature]):
     """
-    TCP fingerprint result
+    TCP fingerprint result.
     """
 
     distance: int = field(init=False)
-    """Estimated distance (TTL)"""
+    """Estimated distance (TTL)."""
 
     def __post_init__(self):
         self.distance = (
@@ -52,7 +53,7 @@ def __post_init__(self):
 
 def guess_distance(ttl: int) -> int:
     """
-    Guess number of hops (distance) for a given packet ttl.
+    Figure out what the TTL distance might have been for a packet.
     """
     return next(
         (initial_ttl - ttl for initial_ttl in (32, 64, 128) if ttl <= initial_ttl),