|
import os |
|
import openai |
|
import json |
|
import rdflib |
|
|
|
|
|
class ExampleGenerator: |
|
def __init__(self): |
|
self.ontologies = {} |
|
self.ontology_files = [] |
|
self.rules = {} |
|
def add_ontology(self, onto): |
|
if onto in self.ontology_files: |
|
raise ValueError("Ontology file already exists.") |
|
else: |
|
onto_data = self.get_ontology_file(onto) |
|
if onto_data: |
|
self.ontology_files.append(onto) |
|
self.ontologies[onto] = self.get_ontology_file(onto) |
|
self.rules[onto] = self.generate_rules(onto) |
|
else: |
|
raise ValueError("Ontology file error.") |
|
def get_ontology_file(self,filename): |
|
text = "" |
|
if os.path.isfile(filename): |
|
with open(filename,'r') as f: |
|
text = f.read() |
|
f.close() |
|
return text |
|
else: |
|
raise ValueError("Invalid filename.") |
|
def ChatGPTTextSplitter(self,text): |
|
"""Splits text in smaller subblocks to feed to the LLM""" |
|
prompt = f"""The total length of content that I want to send you is too large to send in only one piece. |
|
|
|
For sending you that content, I will follow this rule: |
|
|
|
[START PART 1/10] |
|
this is the content of the part 1 out of 10 in total |
|
[END PART 1/10] |
|
|
|
Then you just answer: "Instructions Sent." |
|
|
|
And when I tell you "ALL PARTS SENT", then you can continue processing the data and answering my requests. |
|
""" |
|
if type(text) == str: |
|
textsize = 12000 |
|
blocksize = int(len(text) / textsize) |
|
if blocksize > 0: |
|
yield prompt |
|
|
|
for b in range(1,blocksize+1): |
|
if b < blocksize+1: |
|
prompt = f"""Do not answer yet. This is just another part of the text I want to send you. Just receive and acknowledge as "Part {b}/{blocksize} received" and wait for the next part. |
|
[START PART {b}/{blocksize}] |
|
{text[(b-1)*textsize:b*textsize]} |
|
[END PART {b}/{blocksize}] |
|
Remember not answering yet. Just acknowledge you received this part with the message "Part {b}/{blocksize} received" and wait for the next part. |
|
""" |
|
yield prompt |
|
else: |
|
prompt = f""" |
|
[START PART {b}/{blocksize}] |
|
{text[(b-1)*textsize:b*textsize]} |
|
[END PART {b}/{blocksize}] |
|
ALL PARTS SENT. Now you can continue processing the request. |
|
""" |
|
yield prompt |
|
else: |
|
yield text |
|
elif type(text) == list: |
|
yield prompt |
|
|
|
for n,block in enumerate(text): |
|
if n+1 < len(text): |
|
prompt = f"""Do not answer yet. This is just another part of the text I want to send you. Just receive and acknowledge as "Part {n+1}/{len(text)} received" and wait for the next part. |
|
[START PART {n+1}/{len(text)}] |
|
{text[n]} |
|
[END PART {n+1}/{len(text)}] |
|
Remember not answering yet. Just acknowledge you received this part with the message "Part {n+1}/{len(text)} received" and wait for the next part. |
|
""" |
|
yield prompt |
|
else: |
|
prompt = f""" |
|
[START PART {n+1}/{len(text)}] |
|
{text[n]} |
|
[END PART {n+1}/{len(text)}] |
|
ALL PARTS SENT. Now you can continue processing the request. |
|
""" |
|
yield prompt |
|
|
|
def send_ontology(self): |
|
ontology = "" |
|
if len(self.ontologies) > 0: |
|
for k,v in self.ontologies.items(): |
|
ontology+=v+"\n" |
|
print("Sending Ontology in Parts") |
|
for i in self.ChatGPTTextSplitter(ontology): |
|
print(self.llm_api(i)) |
|
else: |
|
raise ValueError("No loaded ontology to send.") |
|
def llm_api(self,prompt,model="gpt-3.5-turbo"): |
|
messages = [{ |
|
"role":"user", |
|
"content":prompt |
|
}] |
|
res = openai.ChatCompletion.create(model=model,messages=messages,temperature=0) |
|
return res.choices[0].message['content'] |
|
|
|
def generate_rules(self,onto=None): |
|
engagement_actions = ['engagement:Access', |
|
'engagement:Alert', |
|
'engagement:Beacon', |
|
'engagement:Deploy', |
|
'engagement:Obfuscate', |
|
'engagement:Respond' |
|
] |
|
engagement_objects = [ |
|
'engagement:Honeypot', |
|
'engagement:Honeytoken', |
|
'engagement:Breadcrumb', |
|
'engagement:BreadcrumbTrail', |
|
'engagement:LureObject', |
|
'engagement:HoneyObject', |
|
'engagement:Decoy', |
|
'engagement:DataSource' |
|
] |
|
engagement_objectives = [ |
|
'objective:CommandAndControl', |
|
'objective:CredentialAccess', |
|
'objective:DevelopResource', |
|
'objective:Discover', |
|
'objective:EscalatePrivilege', |
|
'objective:Evade', |
|
'objective:Execute', |
|
'objective:Exfilitrate', |
|
'objective:GainInitialAccess', |
|
'objective:Impact', |
|
'objective:MoveLaterally', |
|
'objective:Persist', |
|
'objective:Reconnaissance', |
|
'objective:Affect', |
|
'objective:Collect', |
|
'objective:Detect', |
|
'objective:Direct', |
|
'objective:Disrupt', |
|
'objective:Elicit', |
|
'objective:Expose', |
|
'objective:Motivate', |
|
'objective:Plan', |
|
'objective:Prepare', |
|
'objective:Prevent', |
|
'objective:Reassure', |
|
'objective:Analyze', |
|
'objective:Deny', |
|
'objective:ElicitBehavior', |
|
'objective:Lure', |
|
'objective:TimeSink', |
|
'objective:Track', |
|
'objective:Trap' |
|
] |
|
prefix_ns = {"engagement": "https://ontology.adversaryengagement.org/ae/engagement#", |
|
"objective":"https://ontology.adversaryengagement.org/ae/objective#", |
|
"role":"https://ontology.adversaryengagement.org/ae/role#", |
|
"identity":"https://ontology.adversaryengagement.org/ae/identity#", |
|
"uco-core prefix":"https://ontology.unifiedcyberontology.org/uco/core#", |
|
"uco-types":"https://ontology.unifiedcyberontology.org/uco/types#", |
|
"uco-role":"https://ontology.unifiedcyberontology.org/uco/role#" |
|
} |
|
ns_str ="" |
|
for k,v in prefix_ns.items(): |
|
ns_str+=f"If namespace {k} prefix is used then {v}\n" |
|
lookup = {"1":{"0":['Each'], |
|
"1":['connects to'] |
|
} |
|
} |
|
v = """Remember make a json-ld format example that only uses classes and properties terms from Adversary Engagement Ontology, Unified Cyber Ontology.""" |
|
|
|
structure = {'engagement:Narrative':{'engagement:hasStoryline':{"1":'engagement:Storyline'} |
|
}, |
|
'engagement:Storyline':{'engagement:hasEvent':{"1":'uco-types:Thread'} |
|
}, |
|
'uco-types:Thread':{'co:element':'contains all engagement:PlannedEvents', |
|
'co:item':{"0":'uco-types:ThreadItem one each for each engagement:PlannedEvent'}, |
|
'co:size':"", |
|
'uco-types:threadOriginItem':"is the uco-types:ThreadItem for the first engagement:PlannedEvent", |
|
'uco-types:threadTerminalItem':"is the uco-types:ThreadItem for the last engagement:PlannedEvent" |
|
}, |
|
'co:size':{'@type':'is xsd:nonNegativeInteger', |
|
'@value':"which is the number of uco-types:ThreadItem" |
|
}, |
|
'uco-types:ThreadItem':{'co:itemContent':'is the engagement:PlannedEvent', |
|
'optional uco-types:threadNextItem':"is the next uco-types:ThreadItem for the next engagement:PlannedEvent if there is one", |
|
'optional uco-types:threadPreviousItem':'is the previous uco-types:ThreadItem for the previous' |
|
}, |
|
'engagement:PlannedEvent':{'engagement:eventContext':"connects to one of the following engagement actions:"+"\n\t\t"+"\n\t\t".join(engagement_actions) |
|
}, |
|
'engagement action':{'uco-core:performer':"",'uco-core:object': 'connects to one of the following engagement deception objects'+"\n\t\t"+"\n\t\t".join(engagement_objects) |
|
}, |
|
'engagement deception object':{'engagement:hasCharacterization':{'1':'uco-core:UcoObject'}, |
|
'objective:hasObjective':'with @type objective:Objective and @id with one of the following instances:'+"\n\t\t"+"\n\t\t".join(engagement_objectives), |
|
'uco-core:name':'is the objective' |
|
}, |
|
'person':{'@type':'is uco-identity:Person', |
|
'uco-core:hasFacet':{"1":{'connects to uco-identity:SimpleNameFacet':{'uco-identity:familyName':"",'uco-identity:givenName':""} |
|
} |
|
} |
|
}, |
|
'uco-core:Role':{'@id':"is the role",'uco-core:name': 'is the role' |
|
}, |
|
'uco-core:Role there is a uco-core:Relationship':{'uco-core:kindofRelationship':'is "has_Role"', |
|
'uco-core:source':{"1":"the person who has the role"}, |
|
"uco-core:target":{"1":"uco-core:Role"} |
|
}, |
|
'engagement:BreadcrumbTrail':{'engagement:hasBreadcrumb':{"1":{'uco-types:Thread':{'co:element':"contains all engagement:Breadcrumb that belong to this engagement:BreadcrumbTrail","co:size":"","co:item":"contains all uco-types:ThreadItem one each for each engagement:Breadcrumb","uco-types:threadOriginItem":"is the uco-types:ThreadItem for the first engagement:Breadcrumb belonging to this engagement:BreadcrumbTrail","uco-types:threadTerminalItem":"is the uco-types:ThreadItem for the last engagement:Breadcrumb belonging to this engagement:BreadcrumbTrail"}} |
|
}}, |
|
'engagement:Breadcrumb':{'engagement:hasCharacterization':{"1":{'which connects to a uco-core:UcoObject which':{'uco-core:description':'which describes the object characterizing the breadcrumb'}},'uco-core:name': 'is the role' |
|
}}, |
|
"class":{'@type': 'which is the class', |
|
'@id': 'which is a unique identifier'}, |
|
"ns":ns_str |
|
|
|
} |
|
|
|
def get_list(struct,limiter="\n\t",skippre=False): |
|
all_stat = [] |
|
for k,v in struct.items(): |
|
if k == "ns": |
|
all_stat.append(v) |
|
elif type(v)==dict: |
|
look = "1" |
|
if len(v) > 1: |
|
plural = "has properties:" |
|
else: |
|
plural = "has property:" |
|
if type(v)==dict: |
|
statement = [" ".join([lookup[look]["0"][0],k,plural])] |
|
if skippre: |
|
statement = [" ".join([k,plural])] |
|
for vk, vv in v.items(): |
|
statement.append(limiter) |
|
statement.append(vk) |
|
if type(vv)==dict: |
|
for i in list(lookup.keys()): |
|
if i in vv: |
|
val = v[vk][i] |
|
if type(val)==dict: |
|
statement.append( get_list(val,limiter+"\t",skippre=True) ) |
|
else: |
|
if not skippre: |
|
statement.append(lookup[look]["1"][0]) |
|
statement.append(v[vk][i]) |
|
elif type(vv)==str: |
|
statement.append(v[vk]) |
|
val = " ".join(statement) |
|
all_stat.append(val) |
|
return "\n".join(all_stat) |
|
|
|
v = get_list(structure) |
|
return v |
|
def generate_rule(self,onto=None): |
|
"""Raw rule string of AEO.""" |
|
v = """Remember make a json-ld format example that only uses classes and properties terms from Adversary Engagement Ontology, Unified Cyber Ontology. |
|
|
|
Each engagement:Narrative has property: |
|
engagement:hasStoryline connects to an engagement:Storyline |
|
Each engagement:Storyline has property: |
|
engagement:hasEvent connects to a uco-types:Thread |
|
Each uco-types:Thread has properties: |
|
co:element contains all engagement:PlannedEvents |
|
co:item contains all uco-types:ThreadItem one each for each engagement:PlannedEvent. |
|
co:size |
|
uco-types:threadOriginItem is the uco-types:ThreadItem for the first engagement:PlannedEvent |
|
uco-types:threadTerminalItem is the uco-types:ThreadItem for the last engagement:PlannedEvent |
|
Each co:size has properties: |
|
@type as xsd:nonNegativeInteger |
|
@value which is the number of uco-types:ThreadItem |
|
Each uco-types:ThreadItem has property: |
|
co:itemContent is the engagement:PlannedEvent |
|
optional uco-types:threadNextItem is the next uco-types:ThreadItem for the next engagement:PlannedEvent if there is one, |
|
optional uco-types:threadPreviousItem is the previous uco-types:ThreadItem for the previous engagement:PlannedEvent if there is one |
|
Each engagement:PlannedEvent has property: |
|
engagement:eventContext connects to one of the following engagement actions: |
|
engagement:Access |
|
engagement:Alert |
|
engagement:Beacon |
|
engagement:Deploy |
|
engagement:Obfuscate |
|
engagement:Respond |
|
Each engagement action has properties: |
|
uco-core:performer |
|
uco-core:object which is the object which the action is applied to |
|
Each engagement action has property: |
|
uco-core:object connects to one of the following engagement deception objects: |
|
engagement:Honeypot |
|
engagement:Honeytoken |
|
engagement:Breadcrumb |
|
engagement:BreadcrumbTrail |
|
engagement:LureObject |
|
engagement:HoneyObject |
|
engagement:Decoy |
|
engagement:DataSource |
|
Each engagement deception object has properties: |
|
engagement:hasCharacterization connects to a uco-core:UcoObject |
|
objective:hasObjective with @type objective:Objective and @id with one of the following instances: |
|
objective:CommandAndControl |
|
objective:CredentialAccess |
|
objective:DevelopResource |
|
objective:Discover |
|
objective:EscalatePrivilege |
|
objective:Evade |
|
objective:Execute |
|
objective:Exfilitrate |
|
objective:GainInitialAccess |
|
objective:Impact |
|
objective:MoveLaterally |
|
objective:Persist |
|
objective:Reconnaissance |
|
objective:Affect |
|
objective:Collect |
|
objective:Detect |
|
objective:Direct |
|
objective:Disrupt |
|
objective:Elicit |
|
objective:Expose |
|
objective:Motivate |
|
objective:Plan |
|
objective:Prepare |
|
objective:Prevent |
|
objective:Reassure |
|
objective:Analyze |
|
objective:Deny |
|
objective:ElicitBehavior |
|
objective:Lure |
|
objective:TimeSink |
|
objective:Track |
|
objective:Trap |
|
uco-core:name is the objective |
|
All people have property: |
|
@type is uco-identity:Person |
|
uco-core:hasFacet that connects to one of the following: |
|
uco-identity:SimpleNameFacet which has the property: |
|
uco-identity:familyName |
|
uco-identity:givenName |
|
Each uco-core:Role has properties: |
|
@id is the role |
|
uco-core:name is the role |
|
Each uco-core:Role there is a uco-core:Relationship with properties: |
|
uco-core:kindofRelationship is "has_Role" |
|
uco-core:source connects to the person who has the role |
|
uco-core:target connects to uco-core:Role |
|
Each engagement:BreadcrumbTrail has property: |
|
engagement:hasBreadcrumb connects to uco-types:Thread |
|
This uco-types:Thread has property: |
|
co:element contains all engagement:Breadcrumb that belong to this engagement:BreadcrumbTrail |
|
co:item contains all uco-types:ThreadItem one each for each engagement:Breadcrumb |
|
co:size |
|
uco-types:threadOriginItem is the uco-types:ThreadItem for the first engagement:Breadcrumb belonging to this engagement:BreadcrumbTrail |
|
uco-types:threadTerminalItem is the uco-types:ThreadItem for the last engagement:Breadcrumb belonging to this engagement:BreadcrumbTrail |
|
Each engagement:Breadcrumb has the properties: |
|
engagement:hasCharacterization which connects to a uco-core:UcoObject with the property: |
|
uco-core:description which describes the object characterizing the breadcrumb |
|
All classes must include property: |
|
@type is the class |
|
@id is a unique identifier |
|
|
|
If namespace engagement prefix is used then https://ontology.adversaryengagement.org/ae/engagement# |
|
If namespace objective prefix is used then https://ontology.adversaryengagement.org/ae/objective# |
|
If namespace role prefix is used then https://ontology.adversaryengagement.org/ae/role# |
|
If namespace identity prefix is used then https://ontology.adversaryengagement.org/ae/identity# |
|
If namespace uco-core prefix is used then https://ontology.unifiedcyberontology.org/uco/core# |
|
If namespace uco-types prefix is used then https://ontology.unifiedcyberontology.org/uco/types# |
|
If namespace uco-role prefix is used then https://ontology.unifiedcyberontology.org/uco/role# |
|
""" |
|
return v |
|
|
|
def generate_continue(self): |
|
v = """ |
|
continue |
|
""" |
|
return v |
|
|
|
def raw_prompt(self,description): |
|
|
|
def run(val): |
|
prompt = f"""Give me a full json-ld format example for the following scenario: |
|
{description} |
|
|
|
{"".join(val)} |
|
""" |
|
for i in self.ChatGPTTextSplitter(prompt): |
|
res = self.llm_api(i) |
|
return res |
|
|
|
res_val = run(self.generate_rules()) |
|
try: |
|
val = json.loads(res_val) |
|
return val |
|
except: |
|
|
|
data = [] |
|
data.append(res_val) |
|
while True: |
|
res = self.llm_api(self.generate_continue()) |
|
data.append(res) |
|
try: |
|
full = "".join(data) |
|
return json.loads(full) |
|
except: |
|
pass |
|
|
|
return None |
|
|
|
def check_for_nested(self,jsonObject): |
|
try: |
|
for k,v in jsonObject.items(): |
|
if type(v) == dict: |
|
return True |
|
except: |
|
pass |
|
return False |
|
|
|
def recursive_typelist(self,obj,ls): |
|
if type(obj)!=dict: |
|
return ls |
|
obj_type = obj['@type'].split(":")[0] |
|
if obj_type not in ls: |
|
ls.append(obj_type) |
|
for k,v in obj.items(): |
|
if self.check_for_nested(v): |
|
return_ls = self.recursive_typelist(v,ls) |
|
for re in return_ls: |
|
if re not in ls: |
|
ls.append(re) |
|
return ls |
|
|
|
|
|
def prompt(self,description): |
|
res = self.raw_prompt(description) |
|
|
|
|
|
type_list = [] |
|
try: |
|
for k in res['@graph']: |
|
ns = k['@type'].split(":")[0] |
|
if ns not in type_list: |
|
type_list.append(ns) |
|
type_list = self.recursive_typelist(k,type_list) |
|
new_prefixes = {} |
|
for k,v in res['@context'].items(): |
|
if k in type_list: |
|
new_prefixes[k] = v |
|
res['@context'] = new_prefixes |
|
except: |
|
return res |
|
return res |
|
|
