Welcome to Karton’s documentation!
Karton is a library made for analysis backend orchestration. Allows you to build flexible malware analysis pipelines and attach new Karton Services with ease.
This is achieved by combining powers of a few existing solutions, Karton just glues them together and allows you to have some sane amount of abstraction over them.
Karton ecosystem consists of:
Redis - store used for message exchange between Karton subsystems
Temporary object storage compatible with Amazon S3 API, holds all the heavy objects (aka Resources) like samples, analyses or memory dumps. The recommended one is Silo although Karton should work well with any S3-compatible backend.
Task routing and data exchange is achieved with the help of Karton-System - core of the Karton, which routes the tasks and keeps everything in order (task lifecycle, garbage collection etc.)
Access to the queues is provided by Karton Gateway - a secure Websocket/HTTP-based interface that brokers task queues and resources for Karton services, replacing direct Redis access. Resources are exchanged through S3 presigned URLs.
from karton.core import Karton, Task, Resource
class GenericUnpacker(Karton):
"""
Performs sample unpacking
"""
identity = "karton.generic-unpacker"
filters = [
{
"type": "sample",
"kind": "runnable",
"platform": "win32"
}
]
def process(self, task: Task) -> None:
# Get sample object
packed_sample = task.get_resource('sample')
# Log with self.log
self.log.info(f"Hi {packed_sample.name}, let me analyze you!")
...
# Send our results for further processing or reporting
task = Task(
{
"type": "sample",
"kind": "raw"
}, payload = {
"parent": packed_sample,
"sample": Resource(filename, unpacked)
})
self.send_task(task)
if __name__ == "__main__":
# Here comes the main loop
GenericUnpacker.main()