extras/packetforge/ParseGraph.py

   1 # Copyright (c) 2022 Intel and/or its affiliates.
   2 # Licensed under the Apache License, Version 2.0 (the "License");
   3 # you may not use this file except in compliance with the License.
   4 # You may obtain a copy of the License at:
   5 #
   6 #     http://www.apache.org/licenses/LICENSE-2.0
   7 #
   8 # Unless required by applicable law or agreed to in writing, software
   9 # distributed under the License is distributed on an "AS IS" BASIS,
  10 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  11 # See the License for the specific language governing permissions and
  12 # limitations under the License.
  13
  14 from ProtocolHeader import *
  15 from ForgeResult import *
  16 from Node import *
  17 from Edge import *
  18 import os
  19
  20
  21 class ParseGraph:
  22     def __init__(self):
  23         self.nodeDict = {}
  24         self.edgeDict = {}
  25
  26     def Create(folder):
  27         try:
  28             pg = ParseGraph()
  29             if not os.path.exists(folder):
  30                 print("folder not exisit")
  31                 return None
  32
  33             if os.path.exists(folder + "/nodes"):
  34                 pg.LoadNodesFromDirectory(folder + "/nodes")
  35             if os.path.exists(folder + "/edges"):
  36                 pg.LoadEdgesFromDirectory(folder + "/edges")
  37         except:
  38             print("Failed to create Parse Graph")
  39             return None
  40         else:
  41             return pg
  42
  43     def Nodes(self):
  44         nodes = []
  45         nodes.extend(self.nodeDict.values)
  46         return nodes
  47
  48     def Edges(self):
  49         edges = []
  50         edges.extend(self.edgeDict.values)
  51         return edges
  52
  53     def LoadNodesFromDirectory(self, folder):
  54         for root, dirs, files in os.walk(folder):
  55             for f in files:
  56                 self.LoadNodeFromFile(os.path.join(root, f))
  57
  58     def LoadEdgesFromDirectory(self, folder):
  59         for root, dirs, files in os.walk(folder):
  60             for f in files:
  61                 self.LoadEdgeFromFile(os.path.join(root, f))
  62
  63     def LoadNodeFromFile(self, file):
  64         try:
  65             node = Node.Create(file)
  66
  67             if node == None:
  68                 print("No node created")
  69                 return None
  70
  71             self.AddNode(node)
  72         except:
  73             print("Failed to create node from " + file)
  74
  75     def LoadEdgeFromFile(self, file):
  76         try:
  77             edges = Edge.Create(file)
  78
  79             if edges == None:
  80                 print("No edge created")
  81                 return None
  82
  83             for edge in edges:
  84                 self.AddEdge(edge)
  85         except:
  86             print("Failed to create edge from " + file)
  87
  88     def createProtocolHeader(self, name):
  89         if name in self.nodeDict:
  90             return ProtocolHeader(self.nodeDict[name])
  91         return None
  92
  93     def GetNode(self, name):
  94         if self.nodeDict.has_key(name):
  95             return self.nodeDict[name]
  96         return None
  97
  98     def GetEdge(self, start, end):
  99         key = start + "-" + end
 100         if key in self.edgeDict:
 101             return self.edgeDict[key]
 102         return None
 103
 104     def AddNode(self, node):
 105         if node.Name in self.nodeDict:
 106             print("Warning: node {0} already exist", node.Name)
 107
 108         self.nodeDict[node.Name] = node
 109
 110     def AddEdge(self, edge):
 111         key = edge.Start + "-" + edge.End
 112         if key in self.edgeDict:
 113             print("Warning: edge {0} already exist", key)
 114         self.edgeDict[key] = edge
 115
 116     def Forge(self, path):
 117         headerList = []
 118
 119         # set field value/mask
 120         for headerConfig in path.stack:
 121             header = self.createProtocolHeader(headerConfig.Header)
 122
 123             if header == None:
 124                 return None
 125
 126             for hcf in headerConfig.fields:
 127                 attr = False
 128                 if not header.SetField(hcf.Name, hcf.Value):
 129                     if not header.SetAttribute(hcf.Name, hcf.Value):
 130                         print("failed to set value of " + hcf.Name)
 131                         return None
 132                     else:
 133                         attr = True
 134
 135                 if not attr and not header.SetMask(hcf.Name, hcf.Mask):
 136                     print("failed to set mask of " + hcf.Name)
 137                     return None
 138
 139             header.Adjust()
 140
 141             headerList.append(header)
 142
 143         # apply edge actions and length autoincrease
 144         for i in range(1, len(headerList)):
 145             start = headerList[i - 1]
 146             end = headerList[i]
 147
 148             edge = self.GetEdge(start.Name(), end.Name())
 149
 150             if edge == None:
 151                 print("no edge exist for {0}, {1}", start.Name, end.Name)
 152                 return None
 153
 154             edge.Apply(start, end)
 155
 156             increase = end.GetSize()
 157             for j in range(i):
 158                 headerList[j].AppendAuto(increase)
 159
 160         # resolve buffer
 161         pktLen = 0
 162         for header in headerList:
 163             header.Resolve()
 164             pktLen += len(header.Buffer)
 165
 166         # join buffer
 167         pktbuf = []
 168         mskbuf = []
 169
 170         offset = 0
 171         for header in headerList:
 172             pktbuf.extend(header.Buffer)
 173             mskbuf.extend(header.Mask)
 174
 175             offset += len(header.Buffer)
 176
 177         result = ForgeResult(headerList, pktbuf, mskbuf)
 178
 179         return result