🐍 Unix socket interaction with python

Today it is increasingly easy to integrate third-party components into your software. However, including new dependencies is also risky for your business. This is why you must regularly justify the introduction of a new dependency to your team. We have recently seen with the implementation of multiple factor authentication on the python package repository issue. When the maintainer of the package atomicwrite remove it to protest against a good decision to improve the software supply chain security.

The goal of this post is to demonstrate how to execute a complex interaction without any dependancy, only with the python standard library.

🥅 The goal: list the containers running from the Docker engine

By default, docker engine API is available from a unix socket /var/run/docker.sock but your can open an HTTP interface like that in your systemd config

ExecStart=/usr/bin/dockerd -H fd:// -H tcp://0.0.0.0:2376 --containerd=/run/containerd/containerd.sock

But you can also open a security breach (replacing 0.0.0.0 with localhost is a better idea) and it’s too easy ;)

⚙️ Request on unix socket

In the Python Standard Library, you will with a wonderful package streams

With that function :

asyncio.open_unix_connection("/var/run/docker.sock")

You will be able to write and read HTTP message with that connection, the RFC is here.

HTTP is a super simple protocol based on line of text. You will send a GET request on path /containers/json

And receive several lines (responses headers) a blank line and the content. This one loaded via JSON will give you the result.

So it’s 25 lines of codes to maitain but how many dependancies avoided ? much more …

async def get_containers():
    reader, writer = await asyncio.open_unix_connection("/var/run/docker.sock")
    query = (
        f"GET /containers/json HTTP/1.0\r\n"
        f"\r\n"
    )
    writer.write(query.encode('utf-8'))
    await writer.drain()
    writer.write_eof()
    headers = True
    while headers:
        line = await reader.readline()
        if line == b"\r\n":
            headers = False
        elif not line:
            break
    containers = []
    if not headers:
        data = await reader.readline()
        containers = json.loads(data.decode("utf-8"))
    writer.close()
    await writer.wait_closed()
    return containers
  
containers = asyncio.run(get_containers())

Want to talk about your Sofware Bill Of Material or software supply chain, contact us via our form at https://foss4.eu

Licence: CC BY-NC-ND 4.0