Support LINT, ULINT, LREAL

Author: pjkundert

server/enip/parser.py

@@ -224,12 +224,30 @@ class DINT( TYPE ):
     struct_format		= '<i'
     struct_calcsize		= struct.calcsize( struct_format )
 
+class ULINT( TYPE ):
+    """An EtherNet/IP LINT; 64-bit unsigned integer"""
+    tag_type			= 0x00c9
+    struct_format		= '<Q'
+    struct_calcsize		= struct.calcsize( struct_format )
+
+class LINT( TYPE ):
+    """An EtherNet/IP LINT; 64-bit signed integer"""
+    tag_type			= 0x00c5
+    struct_format		= '<q'
+    struct_calcsize		= struct.calcsize( struct_format )
+
 class REAL( TYPE ):
-    """An EtherNet/IP INT; 32-bit float"""
-    tag_type			= 0x00ca # 202
+    """An EtherNet/IP REAL; 32-bit float"""
+    tag_type			= 0x00ca
     struct_format		= '<f'
     struct_calcsize		= struct.calcsize( struct_format )
 
+class LREAL( TYPE ):
+    """An EtherNet/IP LREAL; 64-bit float"""
+    tag_type			= 0x00cb
+    struct_format		= '<d'
+    struct_calcsize		= struct.calcsize( struct_format )
+
 # Some network byte-order types that are occasionally used in parsing
 class UINT_network( TYPE ):
     """An EtherNet/IP UINT; 16-bit unsigned integer, but in network byte order"""
@@ -1798,12 +1816,32 @@ def __init__( self, name=None, tag_type=None, **kwds ):
                                            destination='.data',	initializer=lambda **kwds: [],
                                                 state=u32d )
 
+        i64d			= octets_noop(	'end64bit',
+                                                terminal=True )
+        i64d[True]	= i64p	= LINT()
+        i64p[None]		= move_if( 	'mov64bit',	source='.LINT',
+                                           destination='.data',	initializer=lambda **kwds: [],
+                                                state=i64d )
+
+        u64d			= octets_noop(	'end64bitu',
+                                                terminal=True )
+        u64d[True]	= u64p	= ULINT()
+        u64p[None]		= move_if( 	'mov64bitu',	source='.ULINT',
+                                           destination='.data',	initializer=lambda **kwds: [],
+                                                state=u64d )
+
         fltd			= octets_noop(	'endfloat',
                                                 terminal=True )
         fltd[True]	= fltp	= REAL()
         fltp[None]		= move_if( 	'movfloat',	source='.REAL', 
                                            destination='.data',	initializer=lambda **kwds: [],
                                                 state=fltd )
+        dltd			= octets_noop(	'enddouble',
+                                                terminal=True )
+        dltd[True]	= dltp	= LREAL()
+        dltp[None]		= move_if( 	'movdouble',	source='.LREAL',
+                                           destination='.data',	initializer=lambda **kwds: [],
+                                                state=dltd )
         # Since a parsed "[S]STRING": { "string": "abc", "length": 3 } is multiple layers deep, and we
         # want to completely eliminate the target container in preparation for the next loop, we'll
         # need to move it up one layer, and then into the final target.
@@ -1846,9 +1884,18 @@ def __init__( self, name=None, tag_type=None, **kwds ):
         slct[None]		= cpppo.decide(	'UDINT',state=u32d,
             predicate=lambda path=None, data=None, **kwds: \
                 UDINT.tag_type == ( data[path+tag_type] if isinstance( tag_type, cpppo.type_str_base ) else tag_type ))
+        slct[None]		= cpppo.decide(	'LINT',	state=i64d,
+            predicate=lambda path=None, data=None, **kwds: \
+                LINT.tag_type == ( data[path+tag_type] if isinstance( tag_type, cpppo.type_str_base ) else tag_type ))
+        slct[None]		= cpppo.decide(	'ULINT',state=u64d,
+            predicate=lambda path=None, data=None, **kwds: \
+                ULINT.tag_type == ( data[path+tag_type] if isinstance( tag_type, cpppo.type_str_base ) else tag_type ))
         slct[None]		= cpppo.decide(	'REAL',	state=fltd,
             predicate=lambda path=None, data=None, **kwds: \
                 REAL.tag_type == ( data[path+tag_type] if isinstance( tag_type, cpppo.type_str_base ) else tag_type ))
+        slct[None]		= cpppo.decide(	'LREAL', state=dltd,
+            predicate=lambda path=None, data=None, **kwds: \
+                LREAL.tag_type == ( data[path+tag_type] if isinstance( tag_type, cpppo.type_str_base ) else tag_type ))
         slct[None]		= cpppo.decide(	'SSTRING', state=sstd,
             predicate=lambda path=None, data=None, **kwds: \
                 SSTRING.tag_type == ( data[path+tag_type] if isinstance( tag_type, cpppo.type_str_base ) else tag_type ))

server/enip_test.py

@@ -12,6 +12,7 @@
 import pytest
 import random
 import socket
+import struct
 import sys
 import traceback
 
@@ -264,6 +265,41 @@ def test_enip_TYPES_numeric():
     assert len( data.typed_data.data ) == 4
     assert data.typed_data.data == [0.0]*4
 
+    # 1 x LREAL
+    pkt				= struct.pack( '<d', 1.23 )
+    data			= cpppo.dotdict()
+    source			= cpppo.chainable( pkt )
+    with enip.typed_data( tag_type=enip.LREAL.tag_type, terminal=True ) as machine:
+        for i,(m,s) in enumerate( machine.run( source=source, data=data )):
+            log.info( "%s #%3d -> %10.10s; next byte %3d: %-10.10r: %r", m.name_centered(),
+                      i, s, source.sent, source.peek(), data )
+        assert i == 15
+    assert len( data.typed_data.data ) == 1
+    assert data.typed_data.data == [1.23]
+
+    # 1 x LINT/ULINT
+    pkt				= struct.pack( '<q', -1234567890123456789 )
+    data			= cpppo.dotdict()
+    source			= cpppo.chainable( pkt )
+    with enip.typed_data( tag_type=enip.LINT.tag_type, terminal=True ) as machine:
+        for i,(m,s) in enumerate( machine.run( source=source, data=data )):
+            log.info( "%s #%3d -> %10.10s; next byte %3d: %-10.10r: %r", m.name_centered(),
+                      i, s, source.sent, source.peek(), data )
+        assert i == 15
+    assert len( data.typed_data.data ) == 1
+    assert data.typed_data.data == [-1234567890123456789]
+
+    pkt				= b'\x00\x00\x00\x00\x00\x00\x00\x80'
+    data			= cpppo.dotdict()
+    source			= cpppo.chainable( pkt )
+    with enip.typed_data( tag_type=enip.ULINT.tag_type, terminal=True ) as machine:
+        for i,(m,s) in enumerate( machine.run( source=source, data=data )):
+            log.info( "%s #%3d -> %10.10s; next byte %3d: %-10.10r: %r", m.name_centered(),
+                      i, s, source.sent, source.peek(), data )
+        assert i == 15
+    assert len( data.typed_data.data ) == 1
+    assert data.typed_data.data == [2**63]
+
 
 # pkt4
 # "4","0.000863000","192.168.222.128","10.220.104.180","ENIP","82","Register Session (Req)"

version.py

@@ -1,2 +1,2 @@
-__version_info__		= ( 4, 3, 0 )
+__version_info__		= ( 4, 3, 1 )
 __version__			= '.'.join( map( str, __version_info__ ))