187 lines
7.1 KiB

One of the four packet-level primitives supported by Geneva. Responsible for any packet-level
modifications (particularly header modifications). It supports the following primitives:
- no operation: it returns the packet given
- replace: it changes a packet field to a fixed value
- corrupt: it changes a packet field to a randomly generated value each time it is run
- add: adds a given value to the value in a field
- compress: performs DNS decompression on the packet (if applicable)
from actions.action import Action
import actions.utils
from layers.dns_layer import DNSLayer
import random
# All supported tamper primitives
SUPPORTED_PRIMITIVES = ["corrupt", "replace", "add", "compress"]
# Tamper primitives we can mutate to by default
ACTIVATED_PRIMITIVES = ["replace", "corrupt"]
class TamperAction(Action):
Defines the TamperAction for Geneva.
frequency = 5
def __init__(self, environment_id=None, field=None, tamper_type=None, tamper_value=None, tamper_proto="TCP"):
Creates a tamper object.
environment_id (str, optional): environment_id of a previously run strategy, used to find packet captures
field (str, optional): field that the object will tamper. If not set, all the parameters are chosen randomly
tamper_type (str, optional): primitive this tamper will use ("corrupt")
tamper_value (str, optional): value to tamper to
tamper_proto (str, optional): protocol we are tampering
Action.__init__(self, "tamper", "both")
self.field = field
self.tamper_value = tamper_value
self.tamper_proto = actions.utils.string_to_protocol(tamper_proto)
self.tamper_proto_str = tamper_proto
self.tamper_type = tamper_type
if not self.tamper_type:
if not self.field:
def mutate(self, environment_id=None):
Mutate can switch between the tamper type, field.
# With some probability switch tamper types
pick = random.random()
if pick < 0.2:
def _mutate_tamper_type(self):
Randomly picks a tamper type to change to.
self.tamper_type = random.choice(ACTIVATED_PRIMITIVES)
if self.tamper_type == "compress":
self.tamper_proto_str = "DNS"
self.tamper_proto = actions.utils.string_to_protocol(self.tamper_proto_str)
self.field = "qd"
def _mutate(self, environment_id):
Mutates this action using:
- previously seen packets with 50% probability
- a fuzzed packet with 50% probability
# Retrieve a new protocol and field options for this protocol
proto, field, value = actions.utils.get_from_fuzzed_or_real_packet(environment_id, 0.5)
self.tamper_proto = proto
self.tamper_proto_str = proto.__name__
self.field = field
self.tamper_value = value
def tamper(self, packet, logger):
Edits a given packet according to the action settings.
# Return packet untouched if not applicable
if not packet.haslayer(self.tamper_proto_str):
return packet
# Retrieve the old value of the field for logging purposes
old_value = packet.get(self.tamper_proto_str, self.field)
new_value = self.tamper_value
# If corrupting the packet field, generate a value for it
if self.tamper_type == "corrupt":
new_value = packet.gen(self.tamper_proto_str, self.field)
elif self.tamper_type == "add":
new_value = int(self.tamper_value) + int(old_value)
elif self.tamper_type == "compress":
return packet.dns_decompress(logger)
except NotImplementedError:
# If a primitive does not support the type of packet given
return packet
except Exception:
# If an unexpected error has occurred
return packet
logger.debug(" - Tampering %s field `%s` (%s) by %s (to %s)" %
(self.tamper_proto_str, self.field, str(old_value), self.tamper_type, str(new_value)))
packet.set(self.tamper_proto_str, self.field, new_value)
return packet
def run(self, packet, logger):
The tamper action runs its tamper procedure on the given packet, and
returns the edited packet down the left branch.
Nothing is returned to the right branch.
return self.tamper(packet, logger), None
def __str__(self):
Defines string representation for this object.
s = Action.__str__(self)
if self.tamper_type == "corrupt":
s += "{%s:%s:%s}" % (self.tamper_proto_str, self.field, self.tamper_type)
elif self.tamper_type in ["replace", "add"]:
s += "{%s:%s:%s:%s}" % (self.tamper_proto_str, self.field, self.tamper_type, self.tamper_value)
elif self.tamper_type == "compress":
s += "{%s:%s:compress}" % ("DNS", "qd", )
return s
def parse(self, string, logger):
Parse out a given string representation of this action and initialize
this action to those parameters.
Note that the given logger is a DIFFERENT logger than the logger passed
to the other functions, and they cannot be used interchangeably. This logger
is attached to the main GA driver, and is run outside the evaluator. When the
action is actually run, it's run within the evaluator, which by necessity must
pass in a different logger.
# Different tamper actions will have different number of parameters
# Count the number of params in this given string
num_parameters = string.count(":")
# If num_parameters is greater than 3, it's not a valid tamper action
if num_parameters > 3 or num_parameters < 2:
msg = "Cannot parse tamper action %s" % string
raise Exception(msg)
params = string.split(":")
if num_parameters == 3:
self.tamper_proto_str, self.field, self.tamper_type, self.tamper_value = params
self.tamper_proto = actions.utils.string_to_protocol(self.tamper_proto_str)
if "options" in self.field:
if not self.tamper_value:
self.tamper_value = '' # An empty string instead of an empty byte literal
# tamper_value might be parsed as a string despite being an integer in most cases.
# Try to parse it out here
if "load" not in self.field:
self.tamper_value = int(self.tamper_value)
self.tamper_proto_str, self.field, self.tamper_type = params
self.tamper_proto = actions.utils.string_to_protocol(self.tamper_proto_str)
return True