add test python

.drone.yml

@@ -5,7 +5,7 @@ name: sec_test
 
 
 steps:
-  - name: security_scan
+  - name: sast_scan
     image: prlab/drone_plugin    
     volumes:
       - name: cache
@@ -15,6 +15,15 @@ steps:
       - echo $(python3 /plugin/main.py)
       - python3 /usr/local/src/scan --src ./src -o /home/reports
 
+  - name: code-analysis
+    image: aosapps/drone-sonar-plugin
+    settings:
+    sonar_host:
+      from_secret: sonar_host
+    sonar_token:
+      from_secret: sonar_token
+
+
   - name: scp
     image: appleboy/drone-scp
     volumes: 

src/parser.py

+import json
+from protocolIE_ID import ProtocolIE_ID  
+
+
+def parse_e2ap_pdu(byte_code):
+    parsed_pdu = {}
+    procedure_code = int.from_bytes(byte_code[0:2], byteorder='big')
+    pdu_length = int.from_bytes(byte_code[2:4], byteorder='big')
+    pdu_data = byte_code[4:4 + pdu_length]
+    initiating_message_offset = 0
+    #procedure_code = int.from_bytes(pdu_data[initiating_message_offset: initiating_message_offset + 1], byteorder='big')
+    criticality = int.from_bytes(pdu_data[initiating_message_offset : initiating_message_offset + 1], byteorder='big')
+    parsed_pdu["InitiatingMessage"] = {
+        "procedureCode": procedure_code,
+        "criticality": criticality,
+        "value": {
+            "protocolIEs":{
+            "ProtocolIE-Container":[]
+            }
+        }
+    }
+
+    ric_indication_offset = initiating_message_offset + 1 
+    protocol_ie_container_length = int.from_bytes(pdu_data[ric_indication_offset: ric_indication_offset + 2], byteorder='big')
+    ric_indication_ie_data = pdu_data[ric_indication_offset + 2:]
+    offset = 0
+    idx = 0
+    while offset < len(ric_indication_ie_data):
+        ie_id = int.from_bytes(ric_indication_ie_data[offset: offset + 2], byteorder='big')        
+        ie_length = int.from_bytes(ric_indication_ie_data[offset + 2: offset + 4], byteorder='big')
+        ie_value = ric_indication_ie_data[offset + 4: offset + 4 + ie_length]
+
+
+        if ie_id == ProtocolIE_ID.ProtocolIE_ID_id_Cause:
+
+            value = ie_value[:ie_length]
+            # need to find some example to extract the field
+            ric_indication_ies = {
+                "Cause": {
+                "id" : ie_id,
+                "criticality" : 0 ,
+                "value" : value.hex().upper()
+            }}
+            parsed_pdu["InitiatingMessage"]["value"]["protocolIEs"]["ProtocolIE-Container"].append(ric_indication_ies)
+
+        # TODO 
+        # check the data field of this IE_ID
+        elif ie_id == ProtocolIE_ID.ProtocolIE_ID_id_CriticalityDiagnostics:
+            ie_criticality = int.from_bytes(ie_value[:1], byteorder='little')
+            msg_length = ie_value[5]
+            
+            msg = ie_value[6:6+msg_length]
+            ric_requestor_id = int.from_bytes(msg[msg_length-4:msg_length-2], byteorder='little')
+            ric_instance_id = int.from_bytes(msg[msg_length-2:], byteorder='little')
+            
+
+            value = ie_value[11:ie_length]
+            ric_indication_ies = {
+                "CriticalityDiagnostics": {
+                "id" : ie_id,
+                "criticality" : 0,
+                "value" : {
+                    "RICRequestID":{
+                        "RICRequestorID": ric_requestor_id,
+                        "RICInstanceID": ric_instance_id
+                    },
+                    "Unknown Field": value.hex().upper()
+                }
+            }}
+            parsed_pdu["InitiatingMessage"]["value"]["protocolIEs"]["ProtocolIE-Container"].append(ric_indication_ies)
+        elif ie_id == ProtocolIE_ID.ProtocolIE_ID_id_GlobalE2node_ID:
+            value = ie_value[:ie_length]
+            ric_indication_ies = {
+                "GlobalE2node_ID": {
+                "id" : ie_id,
+                "criticality" : 0,
+                "value" : value.hex().upper()
+            }}
+            parsed_pdu["InitiatingMessage"]["value"]["protocolIEs"]["ProtocolIE-Container"].append(ric_indication_ies)
+        elif ie_id == ProtocolIE_ID.ProtocolIE_ID_id_GlobalRIC_ID:
+            value = ie_value[:ie_length]
+            ric_indication_ies = {
+                "GlobalRIC_ID": {
+                "id" : ie_id,
+                "criticality" : 0,
+                "value" : value.hex().upper()
+            }}
+            parsed_pdu["InitiatingMessage"]["value"]["protocolIEs"]["ProtocolIE-Container"].append(ric_indication_ies)
+
+
+        #ProtocolIE_ID_id_RANfunctionID : 5 ->  RAN function
+        elif ie_id == ProtocolIE_ID.ProtocolIE_ID_id_RANfunctionID:
+            ric_criticality = int.from_bytes(ie_value[:1], byteorder='big')
+            value = int.from_bytes(ie_value[1:], byteorder='big')
+            ric_indication_ies = { 
+                "RANFunctionID": {
+                "id" : ie_id,
+                "criticality" : ric_criticality ,
+                "value" : value
+            }}
+            parsed_pdu["InitiatingMessage"]["value"]["protocolIEs"]["ProtocolIE-Container"].append(ric_indication_ies)
+           
+        elif ie_id == ProtocolIE_ID.ProtocolIE_ID_id_RANfunctionID_Item:
+            value = ie_value[:ie_length]
+            ric_indication_ies = {
+                "RANFunctionID_Item": {
+                "id" : ie_id,
+                "criticality" : 0,
+                "value" : value.hex().upper()
+            }}
+            parsed_pdu["InitiatingMessage"]["value"]["protocolIEs"]["ProtocolIE-Container"].append(ric_indication_ies)
+
+        elif ie_id == ProtocolIE_ID.ProtocolIE_ID_id_RANfunctionIEcause_Item:
+            value = ie_value[:ie_length]
+            ric_indication_ies = {
+                "RANFunctionIEcause_Item": {
+                "id" : ie_id,
+                "criticality" : 0,
+                "value" : value.hex().upper()
+            }}
+            parsed_pdu["InitiatingMessage"]["value"]["protocolIEs"]["ProtocolIE-Container"].append(ric_indication_ies)
+
+        elif ie_id == ProtocolIE_ID.ProtocolIE_ID_id_RANfunction_Item:
+            value = ie_value[:ie_length]
+            ric_indication_ies = {
+                "RANFunction_Item": {
+                "id" : ie_id,
+                "criticality" : 0,
+                "value" : value.hex().upper()
+            }}
+            parsed_pdu["InitiatingMessage"]["value"]["protocolIEs"]["ProtocolIE-Container"].append(ric_indication_ies)
+
+        elif ie_id == ProtocolIE_ID.ProtocolIE_ID_id_RANfunctionsAccepted:
+            value = ie_value[:ie_length]
+            ric_indication_ies = {
+                "RANFunctionAccepted": {
+                "id" : ie_id,
+                "criticality" : 0,
+                "value" : value.hex().upper()
+            }}
+            parsed_pdu["InitiatingMessage"]["value"]["protocolIEs"]["ProtocolIE-Container"].append(ric_indication_ies)
+        
+        elif ie_id == ProtocolIE_ID.ProtocolIE_ID_id_RANfunctionsAdded:
+            value = ie_value[:ie_length]
+            ric_indication_ies = {
+                "RANFunctionAdded": {
+                "id" : ie_id,
+                "criticality" : 0,
+                "value" : value.hex().upper()
+            }}
+            parsed_pdu["InitiatingMessage"]["value"]["protocolIEs"]["ProtocolIE-Container"].append(ric_indication_ies)
+
+        elif ie_id == ProtocolIE_ID.ProtocolIE_ID_id_RANfunctionsDeleted:
+            value = ie_value[:ie_length]
+            ric_indication_ies = {
+                "RANFunctionDeleted": {
+                "id" : ie_id,
+                "criticality" : 0,
+                "value" : value.hex().upper()
+            }}
+            parsed_pdu["InitiatingMessage"]["value"]["protocolIEs"]["ProtocolIE-Container"].append(ric_indication_ies)
+        
+        elif ie_id == ProtocolIE_ID.ProtocolIE_ID_id_RANfunctionsModified:
+            value = ie_value[:ie_length]
+            ric_indication_ies = {
+                "RANFunctionModified": {
+                "id" : ie_id,
+                "criticality" : 0,
+                "value" : value.hex().upper()
+            }}
+            parsed_pdu["InitiatingMessage"]["value"]["protocolIEs"]["ProtocolIE-Container"].append(ric_indication_ies)
+
+        elif ie_id == ProtocolIE_ID.ProtocolIE_ID_id_RANfunctionsRejected:
+            value = ie_value[:ie_length]
+            ric_indication_ies = {
+                "RANFunctionRejected": {
+                "id" : ie_id,
+                "criticality" : 0,
+                "value" : value.hex().upper()
+            }}
+            parsed_pdu["InitiatingMessage"]["value"]["protocolIEs"]["ProtocolIE-Container"].append(ric_indication_ies)
+
+        elif ie_id == ProtocolIE_ID.ProtocolIE_ID_id_RICaction_Admitted_Item:
+            value = ie_value[:ie_length]
+            ric_indication_ies = {
+                "RICAction_Admitted_Item": {
+                "id" : ie_id,
+                "criticality" : 0,
+                "value" : value.hex().upper()
+            }}
+            parsed_pdu["InitiatingMessage"]["value"]["protocolIEs"]["ProtocolIE-Container"].append(ric_indication_ies)
+
+
+
+        # ProtocolIE_ID_id_RICactionID : 15 
+        elif ie_id == ProtocolIE_ID.ProtocolIE_ID_id_RICactionID:
+            value = int.from_bytes(ie_value[:ie_length], byteorder='big')
+            ric_indication_ies = {
+                "RICactionID":{
+                    "id": ie_id,
+                    "criticality": ric_criticality,
+                    "value": value
+                
+                }
+            }
+            parsed_pdu["InitiatingMessage"]["value"]["protocolIEs"]["ProtocolIE-Container"].append(ric_indication_ies)
+
+
+        elif ie_id == ProtocolIE_ID.ProtocolIE_ID_id_RICaction_NotAdmitted_Item:
+            value = ie_value[:ie_length]
+            ric_indication_ies = {
+                "RICAction_NotAdmitted_Item": {
+                "id" : ie_id,
+                "criticality" : 0,
+                "value" : value.hex().upper()
+            }}
+            parsed_pdu["InitiatingMessage"]["value"]["protocolIEs"]["ProtocolIE-Container"].append(ric_indication_ies)
+        
+        elif ie_id == ProtocolIE_ID.ProtocolIE_ID_id_RICactions_Admitted:
+            value = ie_value[:ie_length]
+            ric_indication_ies = {
+                "RICActions_Admitted": {
+                "id" : ie_id,
+                "criticality" : 0,
+                "value" : value.hex().upper()
+            }}
+            parsed_pdu["InitiatingMessage"]["value"]["protocolIEs"]["ProtocolIE-Container"].append(ric_indication_ies)
+        # 18
+        elif ie_id == ProtocolIE_ID.ProtocolIE_ID_id_RICactions_NotAdmitted:
+            value = ie_value[:ie_length]
+            ric_indication_ies = {
+                "RICActions_NotAdmitted": {
+                "id" : ie_id,
+                "criticality" : 0,
+                "value" : value.hex().upper()
+            }}
+            parsed_pdu["InitiatingMessage"]["value"]["protocolIEs"]["ProtocolIE-Container"].append(ric_indication_ies)
+
+        elif ie_id == ProtocolIE_ID.ProtocolIE_ID_id_RICaction_ToBeSetup_Item:
+            value = ie_value[:ie_length]
+            ric_indication_ies = {
+                "RICAction_ToBeSetup_Item": {
+                "id" : ie_id,
+                "criticality" : 0,
+                "value" : value.hex().upper()
+            }}
+            parsed_pdu["InitiatingMessage"]["value"]["protocolIEs"]["ProtocolIE-Container"].append(ric_indication_ies)
+        
+        # 20
+        elif ie_id == ProtocolIE_ID.ProtocolIE_ID_id_RICcallProcessID:
+            value = ie_value[:ie_length]
+            ric_indication_ies = {
+                "RICCallProcessID": {
+                "id" : ie_id,
+                "criticality" : 0,
+                "value" : value.hex().upper()
+            }}
+            parsed_pdu["InitiatingMessage"]["value"]["protocolIEs"]["ProtocolIE-Container"].append(ric_indication_ies)
+
+        # 21
+        elif ie_id == ProtocolIE_ID.ProtocolIE_ID_id_RICcontrolAckRequest:
+            value = ie_value[:ie_length]
+            ric_indication_ies = {
+                "RICControlAckRequest": {
+                "id" : ie_id,
+                "criticality" : 0,
+                "value" : value.hex().upper()
+            }}
+            parsed_pdu["InitiatingMessage"]["value"]["protocolIEs"]["ProtocolIE-Container"].append(ric_indication_ies)
+
+
+        # 25 
+        elif ie_id == ProtocolIE_ID.ProtocolIE_ID_id_RICindicationHeader:
+            
+            length = int.from_bytes(ie_value[:1], byteorder='big')
+            value = ie_value[1:ie_length]
+            
+
+            ric_indication_ies = {
+                "RICIndicationHeader":{
+                    "id": ie_id,
+                    "criticality": 0,
+                    "value": value.hex().upper()
+
+                }
+            }
+            parsed_pdu["InitiatingMessage"]["value"]["protocolIEs"]["ProtocolIE-Container"].append(ric_indication_ies)
+
+
+        # 26
+        elif ie_id == ProtocolIE_ID.ProtocolIE_ID_id_RICindicationMessage:
+            length = int.from_bytes(ie_value[:1], byteorder='big')
+            value = ie_value[1:ie_length]
+            if len(value) != length:
+                return "{'status':'error','msg':'length is not valid' }" 
+
+            ric_indication_ies = {
+                "RICIndicationMessage":{
+                    "id": ie_id,
+                    "criticality": 0,
+                    "value": str(value)
+
+                }
+            }
+            parsed_pdu["InitiatingMessage"]["value"]["protocolIEs"]["ProtocolIE-Container"].append(ric_indication_ies)
+
+        # 27
+        elif ie_id == ProtocolIE_ID.ProtocolIE_ID_id_RICindicationSN:
+            value = int.from_bytes(ie_value[:ie_length], byteorder='big')
+            ric_indication_ies = {
+                "RICindicationSN":{
+                    "id": ie_id,
+                    "criticality": ric_criticality,
+                    "value": value
+            }}
+            parsed_pdu["InitiatingMessage"]["value"]["protocolIEs"]["ProtocolIE-Container"].append(ric_indication_ies)
+
+        # 28
+        elif ie_id == ProtocolIE_ID.ProtocolIE_ID_id_RICindicationType:
+            value = int.from_bytes(ie_value[:ie_length], byteorder='big')
+            ric_indication_ies = {
+                "RICindicationType":{
+                    "id": ie_id,
+                    "criticality": ric_criticality,
+                    "value": value
+            }}
+            parsed_pdu["InitiatingMessage"]["value"]["protocolIEs"]["ProtocolIE-Container"].append(ric_indication_ies)
+            
+
+        # 29
+        elif ie_id == ProtocolIE_ID.ProtocolIE_ID_id_RICrequestID:
+            ric_criticality = int.from_bytes(ie_value[:1], byteorder='little')
+            ric_requestor_id = int.from_bytes(ie_value[2:], byteorder='little')
+            ric_instance_id = int.from_bytes(ie_value[:2], byteorder='little')
+            ric_indication_ies ={
+                "RICRequestID": {
+                "id" : ie_id, 
+                "criticality": ric_criticality,
+                "value" : {
+                    "RICrequestID":{
+                    "ricRequestorID": ric_requestor_id,
+                    "ricInstanceID": ric_instance_id
+                    }
+                }
+            }}
+            parsed_pdu["InitiatingMessage"]["value"]["protocolIEs"]["ProtocolIE-Container"].append(ric_indication_ies)
+            idx += 1
+        # 30
+        elif ie_id == ProtocolIE_ID.ProtocolIE_ID_id_RICsubscriptionDetails:
+            ric_criticality = 0
+            value = ie_value[:ie_length]
+            ric_indication_ies = {
+                "RICsubscriptionDetails":{
+                    "id": ie_id,
+                    "criticality": ric_criticality,
+                    "value": value.hex().upper()
+            }}
+            parsed_pdu["InitiatingMessage"]["value"]["protocolIEs"]["ProtocolIE-Container"].append(ric_indication_ies)
+
+
+
+        else:
+            print("This ie_id : {} decoder is not implemented".format(ie_id))
+            print("Value : ")
+            print(ie_value)
+        offset += 4 + ie_length
+    return parsed_pdu
+
+
+if __name__ == "__main__":
+    # testing data
+    byte_code = b'\x00\x05\x00R\x00\x00\x06\x00\x1d\x00\x05\x00\x00\x18\x00\x00\x00\x05\x00\x02\x00\x00\x00\x0f\x00\x01\x01\x00\x1c\x00\x01\x00\x00\x19\x00\x04\x03\x13\xf1\x84\x00\x1a\x00*)0,1,2,3,4,5\n1,6,7,8,9,10\n2,11,12,13,14,15'
+    byte_code = b'\x01\x00\x0e\x00\x00\x01\x00\x04\x00\x07\x00\x13\x10\x14\xaa\xcc\xe0'
+    byte_code =  b'\x00\x01\x00\x10\x00\x00\x01\x00\x03\x00\t\x00\x13\xf1\x84P\x00\x03\x04\x01'
+    parsed_pdu = parse_e2ap_pdu(byte_code)
+    print(json.dumps(parsed_pdu))
+

src/protocolIE_ID.py

+
+class ProtocolIE_ID:
+    ProtocolIE_ID_id_Cause = 1
+    ProtocolIE_ID_id_CriticalityDiagnostics = 2
+    ProtocolIE_ID_id_GlobalE2node_ID = 3
+    ProtocolIE_ID_id_GlobalRIC_ID = 4
+    ProtocolIE_ID_id_RANfunctionID = 5
+    ProtocolIE_ID_id_RANfunctionID_Item = 6
+    ProtocolIE_ID_id_RANfunctionIEcause_Item = 7
+    ProtocolIE_ID_id_RANfunction_Item = 8
+    ProtocolIE_ID_id_RANfunctionsAccepted = 9
+    ProtocolIE_ID_id_RANfunctionsAdded = 10
+    ProtocolIE_ID_id_RANfunctionsDeleted = 11
+    ProtocolIE_ID_id_RANfunctionsModified = 12
+    ProtocolIE_ID_id_RANfunctionsRejected = 13
+    ProtocolIE_ID_id_RICaction_Admitted_Item = 14
+    ProtocolIE_ID_id_RICactionID = 15
+    ProtocolIE_ID_id_RICaction_NotAdmitted_Item = 16
+    ProtocolIE_ID_id_RICactions_Admitted = 17
+    ProtocolIE_ID_id_RICactions_NotAdmitted = 18
+    ProtocolIE_ID_id_RICaction_ToBeSetup_Item = 19
+    ProtocolIE_ID_id_RICcallProcessID = 20
+    ProtocolIE_ID_id_RICcontrolAckRequest = 21
+    ProtocolIE_ID_id_RICcontrolHeader = 22
+    ProtocolIE_ID_id_RICcontrolMessage = 23
+    ProtocolIE_ID_id_RICcontrolStatus = 24
+    ProtocolIE_ID_id_RICindicationHeader = 25
+    ProtocolIE_ID_id_RICindicationMessage = 26
+    ProtocolIE_ID_id_RICindicationSN = 27
+    ProtocolIE_ID_id_RICindicationType = 28
+    ProtocolIE_ID_id_RICrequestID = 29
+    ProtocolIE_ID_id_RICsubscriptionDetails = 30
+    ProtocolIE_ID_id_TimeToWait = 31
+    ProtocolIE_ID_id_RICcontrolOutcome = 32
+