DeseriGuard/policygenerator.py at master · THU-WingTecher/DeseriGuard · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
import os
import json
import argparse


checkerPosition_jdk7 = {
    "java/io/ObjectInputStream:readObject" : "->java/io/ObjectInputStream:readOrdinaryObject:1774->java/io/ObjectStreamClass:forClass",
    "org/apache/ofbiz/base/util/SafeObjectInputStream:readObject" : "->java/io/ObjectInputStream:readOrdinaryObject:1774->java/io/ObjectStreamClass:forClass",
    "java/beans/XMLDecoder:readObject" : "->com/sun/beans/decoder/NewElementHandler:addAttribute:25->com/sun/beans/decoder/DocumentHandler:findClass",
    "com/thoughtworks/xstream/XStream:fromXML" : "->com/sun/beans/decoder/NewElementHandler:addAttribute:25->com/sun/beans/decoder/DocumentHandler:findClass"
}

checkerPosition_jdk8 = {
    "java/io/ObjectInputStream:readObject" : "->java/io/ObjectInputStream:readOrdinaryObject:2189->java/io/ObjectStreamClass:forClass",
    "org/apache/ofbiz/base/util/SafeObjectInputStream:readObject" : "->java/io/ObjectInputStream:readOrdinaryObject:2189->java/io/ObjectStreamClass:forClass",
    "java/beans/XMLDecoder:readObject" : "->com/sun/beans/decoder/NewElementHandler:addAttribute:25->com/sun/beans/decoder/DocumentHandler:findClass",
    "com/thoughtworks/xstream/XStream:fromXML" : "->com/sun/beans/decoder/NewElementHandler:addAttribute:25->com/sun/beans/decoder/DocumentHandler:findClass",
}


def format(str):
    str = str.replace("[", "")
    str = str.replace("\\", "")
    if str.startswith("L"):
        str = str[1:]
    str = str.replace(";", "")
    # str = str.replace("/", ".")
    return str

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--path", "-p", required=True, help="json file generated by codeql")
    parser.add_argument("--jdk", "-j", required=False, help="jdk version", default=8, type=int)

    args = parser.parse_args()

    jsonPath = args.path

    with open(jsonPath) as f:
        jsonFile = json.load(f)
    results = jsonFile["runs"][0]["results"]
    policy = open("policy", "w")
    policyMap = {}
    for res in results:

        deseraLocation = res["codeFlows"][0]["threadFlows"][0]["locations"][0]["location"]
        file = deseraLocation["physicalLocation"]["artifactLocation"]["uri"]
        line = deseraLocation["physicalLocation"]["region"]["startLine"]
        message = res["message"]["text"].split("!")
        print(message)
        targetType = message[3]
        # if targetType.startswith("java/lang/"):
        #     continue
        function = message[5]
        objectName = format(message[7])
        deserMethodObject = format(message[9])
        deserMethod = message[11]
        if args.jdk == 8:
            checker = checkerPosition_jdk8[deserMethodObject+":" + deserMethod]
        else:
            checker = checkerPosition_jdk7[deserMethodObject+":" + deserMethod]

        readObjectLocation = file+":"+objectName+":"+function +":"+str(line) + "->" + deserMethodObject+":" + deserMethod + checker
        if readObjectLocation not in policyMap:
            policyMap[readObjectLocation] = set()
        policyMap[readObjectLocation].add(format(targetType))


    for loca, types in policyMap.items():
        policy.write(loca+"\n")
        for type in types:
            if len(type)==1:
                continue
            if type[-2] == '/':
                continue
            policy.write(type+"\n")
    policy.close()