# Ziirish's Home :: Blog

Ziirish's Pub

## Python __getattribute__ magic

While working on Burp-UI some weeks ago, I did some refactoring using the magic getattribute method.

I did not find a lot of examples on the Internet because it is usually not recommended to play with this function, but I think the use-case I'm about to show you is interesting.

# Introduction

Burp-UI allows you to run several agents if you have several Burp servers and all these agents will be available through one unique UI.

The following schema shows you how it works:

Every agent must implement the BUIbackend interface, but in fact an Agent is just a basic server acting between the actual backend and the UI engine.

It reads the commands sent by the UI engine and pass them to the backend. This way, you can have both Burp 2 and Burp 1 servers all centralized in the same dashboard.

On the other side, the UI engine also have a special backend to interact with the good agent. This backend also have to respect the interface.

If we do the math, everytime I wanted to add a new feature I had to implement 4 functions: one per backend + the agent. That's a lot of work. So I decided to cleanup a bit this mess and here is the result.

The special UI backend is just a proxy between the UI and the agent. There is no business code in it. The agent is also here to interact between the special remote backend and the actual local backend that to the business. The Burp 2 backend inherit the Burp 1 one because both of them share some code. So if we redo the math at this point, we see we can almost divide the efforts by 4 because in most cases, only the Burp 1 backend will carry the business code.

# The Agent

Indeed, the Agent code looked like this:

class BUIAgent:
def __init__(self, conf=None, debug=False):
# some code [...]
module = 'burpui.misc.backend.burp{0}'.format(self.vers)
try:
mod = __import__(module, fromlist=['Burp'])
Client = mod.Burp
self.backend = Client(conf=conf)
except Exception, e:
sys.exit(2)

self.methods = {
'status': self.backend.status,
'get_backup_logs': self.backend.get_backup_logs,
'get_counters': self.backend.get_counters,
'is_backup_running': self.backend.is_backup_running,
'is_one_backup_running': self.backend.is_one_backup_running,
'get_all_clients': self.backend.get_all_clients,
'get_client': self.backend.get_client,
'get_tree': self.backend.get_tree,
'restore_files': self.backend.restore_files,
'store_conf': self.backend.store_conf,
'get_parser_attr': self.backend.get_parser_attr
}

self.server = AgentServer((self.bind, self.port), AgentTCPHandler, self)

class AgentTCPHandler(SocketServer.BaseRequestHandler):
"One instance per connection.  Override handle(self) to customize action."
def handle(self):
# self.request is the client connection
try:
r, _, _ = select.select([self.request], [], [], 5)
if not r:
raise Exception ('Socket timed-out')
lengthbuf = self.request.recv(8)
length, = struct.unpack('!Q', lengthbuf)
data = self.recvall(length)
_, w, _ = select.select([], [self.request], [], 5)
if not w:
raise Exception ('Socket timed-out')
if j['func'] not in self.server.agent.methods:
self.request.sendall('KO')
return
self.request.sendall('OK')
if j['args']:
res = json.dumps(self.server.agent.methods[j['func']](**j['args']))
else:
res = json.dumps(self.server.agent.methods[j['func']]())

As you can see, the Agent just keeps a list of methods and associate each one of them with the backend's function. It means anytime I wanted to add a new function, I had to update this list...

And here comes the magic! The python abc module allows us to define abstract base classes. So I used this module to define the BUIbackend interface. Now if we want to inherit the BUIbackend abstract class, we must implement its abstract methods. Thanks to this, we know exactly what methods are implemented by any class inheriting the interface.

The new agent code now looks like this:

class BurpHandler(BUIbackend):
# These functions MUST be implemented because we inherit an abstract class.
# The hack here is to get the list of the functions and let the interpreter
# think we don't have to implement them.
# Thanks to this list, we know what function are implemented by our backend.
foreign = BUIbackend.__abstractmethods__
BUIbackend.__abstractmethods__ = frozenset()

def __init__(self, vers=1, conf=None):
self.vers = vers

module = 'burpui.misc.backend.burp{0}'.format(self.vers)
try:
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
mod = __import__(module, fromlist=['Burp'])
Client = mod.Burp
self.backend = Client(conf=conf)
except Exception as e:
sys.exit(2)

def __getattribute__(self, name):
# always return this value because we need it and if we don't do that
# we'll end up with an infinite loop
if name == 'foreign':
return object.__getattribute__(self, name)
# now we can retrieve the 'foreign' list and know if the object called
# is in the backend
if name in self.foreign:
return getattr(self.backend, name)
return object.__getattribute__(self, name)

And now the handle code calls the methods like this: res = json.dumps(getattr(self.cli, j['func'])(**j['args']))

Pretty nice, no?

# The multi backend

The special backend acting as a proxy between the UI and the agent is called multi and it looked like this:

class Burp(BUIbackend):

def __init__(self, app=None, conf=None):
self.app = app
self.servers = {}
self.app.config['SERVERS'] = []
self.running = {}
if conf:
config = ConfigParser.ConfigParser()
with open(conf) as fp:
for sec in config.sections():
r = re.match('^Agent:(.+)$', sec) if r: try: host = config.get(sec, 'host') port = config.getint(sec, 'port') password = config.get(sec, 'password') ssl = config.getboolean(sec, 'ssl') except Exception, e: self.app.logger.error(str(e)) self.servers[r.group(1)] = NClient(app, host, port, password, ssl) self.app.logger.debug(self.servers) for key, serv in self.servers.iteritems(): self.app.config['SERVERS'].append(key) """ Utilities functions """ def status(self, query='\n', agent=None): """ status connects to the burp status port, ask the given 'question' and parses the output in an array """ return self.servers[agent].status(query) def get_backup_logs(self, n, c, forward=False, agent=None): """ parse_backup_log parses the log.gz of a given backup and returns a dict containing different stats used to render the charts in the reporting view """ return self.servers[agent].get_backup_logs(n, c, forward) """ And so on... """ As you can see, this backend creates a NClient object per agent found in the configuration and it implements every method provided by the interface. You may also notice the code of each methods looks very similar. Now what about the NClient object? Well, here it is (or was, actually): class NClient(BUIbackend): def __init__(self, app=None, host=None, port=None, password=None, ssl=None): self.host = host self.port = port self.password = password self.ssl = ssl self.connected = False self.app = app """ Some network related code """ def do_command(self, data=None): self.conn() res = '[]' if not data: return res try: data['password'] = self.password raw = json.dumps(data) length = len(raw) self.sock.sendall(struct.pack('!Q', length)) self.sock.sendall(raw) r, _, _ = select.select([self.sock], [], [], 5) if not r: raise Exception ('Socket timed-out') tmp = self.sock.recv(2) if 'OK' != tmp: return res r, _, _ = select.select([self.sock], [], [], 5) if not r: raise Exception ('Socket timed-out') lengthbuf = self.sock.recv(8) length, = struct.unpack('!Q', lengthbuf) r, _, _ = select.select([self.sock], [], [], 5) if not r: raise Exception ('Socket timed-out') res = self.sock.recv(length) except Exception, e: self.app.logger.error(str(e)) finally: self.close() return res """ Utilities functions """ def status(self, query='\n', agent=None): """ status connects to the burp status port, ask the given 'question' and parses the output in an array """ data = {'func': 'status', 'args': {'query': query}} return json.loads(self.do_command(data)) def get_backup_logs(self, n, c, forward=False, agent=None): """ parse_backup_log parses the log.gz of a given backup and returns a dict containing different stats used to render the charts in the reporting view """ data = {'func': 'get_backup_logs', 'args': {'n': n, 'c': c, 'forward': forward}} return json.loads(self.do_command(data)) Again, the code of each backend method is kinda boring because we just need to encode its arguments in a json and pass them through a socket. The new code is sexier imho: INTERFACE_METHODS = BUIbackend.__abstractmethods__ class ProxyCall(object): """Class to dispatch call of unknown methods in order to dynamically call agents one without maintaining the explicit list of methods. """ def __init__(self, proxy, method, network=False): self.proxy = proxy self.method = method self.network = network def __call__(self, *args, **kwargs): """This is were the proxy call (and the magic) occurs""" # retrieve the original function prototype proto = getattr(BUIbackend, self.method) args_name = list(proto.__code__.co_varnames) # skip self args_name.pop(0) # we transform unnamed arguments to named ones # example: # def my_function(toto, tata=None, titi=None): # # x = my_function('blah', titi='blih') # # => {'toto': 'blah', 'titi': 'blih'} encoded_args = {} for idx, opt in enumerate(args): encoded_args[args_name[idx]] = opt encoded_args.update(kwargs) # Special case for network calls if self.network: data = {'func': self.method, 'args': encoded_args} if self.method == 'restore_files': return self.proxy.do_command(data) return json.loads(self.proxy.do_command(data)) # normal case for "standard" interface if 'agent' not in encoded_args: raise AttributeError(str(encoded_args)) agentName = encoded_args['agent'] # we don't need this argument anymore del encoded_args['agent'] try: agent = self.proxy.servers[agentName] except KeyError: # This exception should be forwarded to the final user if not agentName: msg = "You must provide an agent name" else: msg = "Agent '{}' not found".format(agentName) raise BUIserverException(msg) return getattr(agent, self.method)(**encoded_args) class NClient(BUIbackend): # These functions MUST be implemented because we inherit an abstract class. # The hack here is to get the list of the functions and let the interpreter # think we don't have to implement them. # Thanks to this list, we know what function are implemented by our backend. foreign = INTERFACE_METHODS BUIbackend.__abstractmethods__ = frozenset() def __init__(self, app=None, host=None, port=None, password=None, ssl=None, timeout=5): self.host = host self.port = port self.password = password self.ssl = ssl self.connected = False self.app = app self.timeout = timeout or 5 def __getattribute__(self, name): # always return this value because we need it and if we don't do that # we'll end up with an infinite loop if name == 'foreign': return object.__getattribute__(self, name) # now we can retrieve the 'foreign' list and know if the object called # needs a dynamic implementation if name in self.foreign: return ProxyCall(self, name, network=True) return object.__getattribute__(self, name) Now this is getting pretty convenient because every backend method is dynamically implemented thanks to the __getattribute__ method. But what if I want to overwrite a method for some reason? Well, using a custom decorator, we can and here is what it looks like: def implement(func): """A decorator indicating the method is implemented. For the agent and the 'multi' backend, we inherit the backend interface but we don't really implement it because we just act as a proxy. But maintaining the exhaustive list of methods in several places to always implement the same "proxy" thing was painful so I ended up cheating to dynamically implement those methods thanks to the __getattribute__ magic function. But sometimes we want to implement specific things, hence this decorator to indicate we don't want the default "magic" implementation and use the custom implementation instead. """ func.__ismethodimplemented__ = True return func class Burp(BUIbackend): """The :class:burpui.misc.backend.multi.Burp class provides a consistent backend to interact with agents. It is actually the *real* multi backend implementing the :class:burpui.misc.backend.interface.BUIbackend class. For each agent found in the configuration, it will load a :class:burpui.misc.backend.multi.NClient class. :param server: Burp-UI server instance in order to access logger and/or some global settings :type server: :class:burpui.server.BUIServer :param conf: Configuration file to use :type conf: str """ # These functions MUST be implemented because we inherit an abstract class. # The hack here is to get the list of the functions and let the interpreter # think we don't have to implement them. # Thanks to this list, we know what function are implemented by our backend. foreign = INTERFACE_METHODS BUIbackend.__abstractmethods__ = frozenset() def __init__(self, server=None, conf=None): """ :param server: Application context :type server: :class:burpui.server.BUIServer """ self.app = server self.acl_handler = server.acl_handler self.servers = {} self.app.config['SERVERS'] = [] self.running = {} if conf: config = ConfigParser.ConfigParser() with open(conf) as fp: config.readfp(fp) for sec in config.sections(): r = re.match('^Agent:(.+)$', sec)
if r:
ssl = False
host = self._safe_config_get(config.get, 'host', sec)
port = self._safe_config_get(config.getint, 'port', sec, cast=int)
ssl = self._safe_config_get(config.getboolean, 'ssl', sec, cast=bool)
timeout = self._safe_config_get(config.getint, 'timeout', sec, cast=int)

self.servers[r.group(1)] = NClient(self.app, host, port, password, ssl, timeout)

self.app.config['SERVERS'] = [ x for x in viewkeys(self.servers) ]

def __getattribute__(self, name):
# always return this value because we need it and if we don't do that
# we'll end up with an infinite loop
if name == 'foreign':
return object.__getattribute__(self, name)
# now we can retrieve the 'foreign' list and know if the object called
# needs to be "proxyfied"
if name in self.foreign:
proxy = True
func = None
try:
func = object.__getattribute__(self, name)
proxy = not getattr(func, '__ismethodimplemented__', False)
except:
pass
self.logger.debug('func: {} - {}'.format(name, proxy))
if proxy:
return ProxyCall(self, name)
elif func:
return func
return object.__getattribute__(self, name)

@implement
def is_one_backup_running(self, agent=None):
"""See :func:burpui.misc.backend.interface.BUIbackend.is_one_backup_running"""
r = []
if agent:
r = self.servers[agent].is_one_backup_running(agent)
self.running[agent] = r
else:
r = self._backup_running_parallel()

self.running = r
self.refresh = time.time()
return r

Every method decorated with the implement decorator will be returned as is by the magic __getattribute__ function. The others will be dynamically generated!

# Conclusion

I hope you now see what magic means when talking about python magic methods