Changeset 4858

Timestamp:

05/31/09 01:36:46 (4 years ago)

Author:

getxsick

Message:

Implementation of umpa.sniffing.from_file()

You can get list of packets, also call callback func, set filters etc.
This function is not tested yet. to_file() should be implemented first ;)

Also a change in libpcap abstract.
No more open_offline. open_online renamed to open_pcap which handles with both cases.

Location:

branch/UMPA

Files:

• tests/unit/test_sniffing/test_libpcap/test_pypcap.py (modified) (3 diffs)
• tests/unit/test_sniffing/test_sniffing_init.py (modified) (1 diff)
• umpa/sniffing/__init__.py (modified) (4 diffs)
• umpa/sniffing/libpcap/_abstract.py (modified) (1 diff)
• umpa/sniffing/libpcap/pypcap.py (modified) (1 diff)

branch/UMPA/tests/unit/test_sniffing/test_libpcap/test_pypcap.py

@@ -42 +42 @@
     def test_openlive(self):
         try:
-            obj = pypcap.open_live()
+            obj = pypcap.open_pcap()
             assert obj.device == pcap.lookupdev()
-            obj = pypcap.open_live(device="any") # XXX can we use 'any'?
+            obj = pypcap.open_pcap(device="any") # XXX can we use 'any'?
             assert obj.device == "any"
         except UMPASniffingException:
@@ -63 +63 @@
 
         try:
-            p = pypcap.open_live("any", to_ms=100)
+            p = pypcap.open_pcap("any", to_ms=100)
             p.setfilter("src host 1.2.3.4 and src port 99")
             th = SendPacket(umpa.Packet(IP(source_address="1.2.3.4"),
@@ -88 +88 @@
         th.start()
         try:
-            p = pypcap.open_live("any", to_ms=100)
+            p = pypcap.open_pcap("any", to_ms=100)
             p.setfilter("src host 1.2.3.4 and src port 99")
             for i in xrange(amount):

branch/UMPA/tests/unit/test_sniffing/test_sniffing_init.py

@@ -90 +90 @@
                         callback_args=[2,1] )
         th.join()
+
+    def test_from_file(self):
+        py.test.skip("implement to_file first")

branch/UMPA/umpa/sniffing/__init__.py

@@ -19 +19 @@
 # along with this library; if not, write to the Free Software Foundation, 
 # Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA 
+
+import os.path
 
 import umpa
@@ -73 +75 @@
     """
 
-    session = lpcap.open_live(device, snaplen, promisc, timeout)
+    session = lpcap.open_pcap(device, snaplen, promisc, timeout)
     if filter:
         session.setfilter(filter)
@@ -134 +136 @@
     """
 
-    session = lpcap.open_live(device, snaplen, promisc, timeout)
+    session = lpcap.open_pcap(device, snaplen, promisc, timeout)
     if filter:
         session.setfilter(filter)
@@ -146 +148 @@
     return sniff(1, device='any')
 
-def from_file():
-    pass
+def from_file(filename, count=0, filter=None, callback=None,
+                                                        callback_args=None):
+    """
+    Load data from pcap file instead of sniffing online.
+
+    Call callback for each or return list of packets.
+
+    @type filename: C{str}
+    @param filename: path to a file in pcap format
+
+    @type count: C{int}
+    @param count: number of sniffing packets; 0 means infinity (default: I{0})
+
+    @type filter: C{str}
+    @param filter: BPF filter
+
+    @type callback: C{func}
+    @param callback: function with (timestamp, pkt, *callback_args) prototype
+
+    @type callback_args: C{list}
+    @param callback_args: additional arguments for callback function
+    """
+
+    if os.path.isfile(filename):
+        f = lpcap.open_pcap(filename)
+    else:
+        raise UMPASniffingException("can't open file: %s" % filename)
+
+    if filter:
+        f.setfilter(filter)
+    if callback is not None:
+        f.loop(count, callback, *callback_args)
+    elif count > 0:
+        packets = []
+        for i in xrange(count):
+            packets.append(f.next())
+    else:
+        packets = f.readpkts()
+
+    return packets
 
 def to_file():

branch/UMPA/umpa/sniffing/libpcap/_abstract.py

@@ -48 +48 @@
                     "selected libpcap backend or abstract module")
 
-def open_offline(fname):
-    """
-    Open a file in tcpdump format for reading.
-    """
-
-    raise NotImplementedError("not implemented method for the "
-                    "selected libpcap backend or abstract module")
-
-class open_live(object):
+class open_pcap(object):
     """
     Packet capture descriptor.
+
+    This is for online and offline capturing.
     """
 

branch/UMPA/umpa/sniffing/libpcap/pypcap.py

@@ -41 +41 @@
     return result
 
-class open_live(open_live):
+class open_pcap(open_pcap):
     def __init__(self, device=None, snaplen=1024, promisc=True, to_ms=0):
         if device is None: