Konnekt CLI - A Look Inside

Introduction

My first foray into the world of Python networking is my currently ongoing project, Konnekt.

It has been quite interesting to work on, as I have integrated certain technologies that I had relatively little experience with. Konnekt is a lightweight command-line application written in Python that collects TCP connections and UDP socket endpoints, displaying them alongside connection statistics in formatted terminal tables using the Rich library.

Rich, whose table API I utilized, allows for human-readable data to be rendered into a variety of structures, including tables.

Of course, I have greater plans for the project, and, as of the publication of this article, it is currently still in active development at v0.2.0.

What I Learned

I have endeavored to glean as much as I can from the project, particularly identifying connections by process ID (PID) and creating a deduplication algorithm that removes duplicates from the rendered table. The deduplication algorithm is quite simple, relying upon a "best effort" methodology, as it is currently unable to perfectly remove all duplicates while preserving unique connections. This is due to an issue with the file descriptors (FDs) and the aforementioned PIDs. FDs can be reused multiple times by multiple processes, which leads to an unusual scenario where even fallback identifiers cannot properly filter the sockets. This was a learning experience in the art of documentation, something considered a ubiquitous companion to any professionally made program, as I outlined the limitations of my algorithm. I chose PIDs and FDs as my first-line filter because they're generally essential to each socket, denoting the process and file they belong to, making them the most informative values of a socket.

However, there are instances where these values are unavailable, which led me to create fallback statements that filter based on the socket IPs and ports. When records lack the specific values required for the first pass—the FD and PID—the algorithm uses a fallback statement wherein it checks for the IPs and ports of both local and remote addresses. If a sufficient set of values can't be constructed from the given record, it is excluded from the table.

Here I have demonstrated the deduplication algorithm as of this point in time:

def deduplicate_connections(record):
    seen_connections = set()
    accepted_records = []

    for i in record:
        connection = identify_connection(i)

        if connection is None:
            continue

        if connection not in seen_connections:
            accepted_records.append(i)
            seen_connections.add(connection)

    return accepted_records

The circumstances in which these values are truly unique yet treated as duplicates are uncommon, yet still worth discussing. This occurs because FDs are used to describe a descriptor within a process defined by a PID. Because FDs are process-local, the same FD can be used without conflict. Although FDs can be used to identify particular resources and individual processes, they are not globally unique. Different processes may utilize the same FD, even when neither process has terminated. Consequently, records constructed from incomplete socket metadata cannot perfectly identify uniquely distinguished records.

There are really two scenarios in which this limitation is best demonstrated:

• Identity Collision: Different sockets may provide the same identifier, resulting in a unique record being discarded.
• Incomplete Records: Some records may lack the PID or FD values required for the algorithm.

The current implementation separates socket identification from deduplication. The identify_connection() function serves to construct an identifier from the available socket information, while deduplicate_connections() maintains a set of previously seen identifiers. Each record with a valid identifier is retained only if its identifier has not already been encountered before.

CI and Testing

In order to create a testing suite for the program, I utilized pytest, directly running it from the CLI at first. Pytest is a testing framework commonly used in Python programming for the purposes of testing modular algorithms. This was also the point in time where I began to make the program more modular, introducing separations of responsibility among the functions in the program. At first, I created the identify_connection() function, then wrote almost ten tests for it in a suite utilizing both singular test functions and a function that utilized parameterization in order to test multiple ports. The identification function was then utilized in the deduplication algorithm to compare connections in a set. A set was utilized in this function for comparison because it provides average-case constant-time membership checks, allowing the algorithm to determine if a record has already been encountered. Modularizing the project allowed me to test individual components independently, rather than relying on tests against a single, tightly coupled codebase.

After tests were written for the two major algorithms, a CI pipeline was set up so that testing and formatting could be checked whenever a push to the repository occurred. The runner for the CI pipeline is both self-hosted and self-managed, utilizing the aforementioned pytest and Ruff, a Python linter and formatter. The runner, specifically made and maintained for the purposes of only the Konnekt repository, is run on a Debian 13 instance virtualized on my Proxmox VE, which runs on my home network. The runner utilizes Forgejo Actions, the CI system Codeberg provides for the purposes of testing, and Forgejo Runner, the system that actually executes code on the remote Debian system. Podman was chosen in order to run the system as a container under a less-privileged user, which restricts the execution of the code to a specific filesystem associated with that user.

The reason I chose to self-host the Forgejo runner instead of using standard Codeberg runners was that I wanted both to learn how to do so for future purposes and to have complete control over the execution of my code for testing. I had relatively little experience with Forgejo's CI pipelines before this and wanted to implement a continuous integration system and, eventually, a deployment system for this project, as well as future ones. When setting up the runner, I learned quite a bit about Linux user management, as I had not managed users at that level before, expanding my horizons on what's truly possible in systems administration. I had trouble initially with the codes (e.g., 755) utilized for restricting read/write abilities, but eventually got the hang of it, despite it being one of the most surprisingly difficult parts of the setup.

Challenges and Lessons Learned

When implementing the modularization of the program, I tended to have the most challenges. Separating identification from the nominal, full program in a single file was difficult, as extracting parts of what was initially one large function tended towards the challenging side. Separating the two algorithms required some rewriting of the code, especially in setting up the required parameters for the function. Then, implementing the identification function in the deduplication algorithm was where it became slightly easier. The deduplication algorithm, demonstrated above, was simple due to its reliance on the modular code, allowing for consistent readability across the program. I learned that it is much easier to work on code when it is broken into smaller pieces, as modularity not only benefits testing, as previously mentioned, but also improves the architect's ability to work with the respective pieces.

There were some testing-related challenges as well, as creating mock socket records was accomplished by using SimpleNamespace from the "types" module. These allowed for labeled variables to be initialized within the namespace, and then registered by the respective algorithms, which used the object's values for identification and deduplication. Creating edge tests proved to be the most challenging. Creating them, as well as having to adapt to failure, was certainly a learning experience in both the writing of algorithms and tests. Developing edge-case tests proved particularly challenging, as they required me to anticipate unusual inputs and reconsider assumptions made during implementation. When tests failed, I had to determine whether the problem originated in the test itself or in the underlying algorithm.

This is an example of a basic test run on the identification algorithm:

    def test_records_correct(self):
        laddr = types.SimpleNamespace(ip="1.1.1.1", port="102")
        test_record = types.SimpleNamespace(
            pid=103, fd=102, family="AF_INET", laddr=laddr
        )
        assert identify_connection(test_record) == (
            "PID and FD",
            test_record.pid,
            test_record.fd,
            test_record.family,
        )

Finally, a very specific lesson learned was tuple unpacking, something I had relatively little knowledge of before referencing Slatkin's Effective Python during a study day between work sessions. Slatkin's explanation, as well as his demonstrations, helped me significantly in writing the final code in the main.py orchestration module of the code. In this module, I also implemented the Rich table API, allowing me to generate a table with true human-readability, followed later by a secondary table rendering connection statistics. Using tuple unpacking, I was able to receive multiple tables from the API and print them using Rich's "Console" object and its print method. It should be noted that this is a Python feature and not a Rich feature, wherein Rich tables were unpacked for printing.

Here is an example of the tuple unpacking used in the program entry point:

if __name__ == "__main__":
    console = Console()
    table, stats = main()
    console.print(table, stats)

Overall, these experiences reinforced the importance of designing software with testability and maintainability in mind, rather than treating testing and automation as separate concerns introduced only after developing the final, stable version.

Looking Ahead

The project's current outlook is to create a stable version by version 5, which will be known as v1.0.0. In this stable version, the user will not only be able to collect snapshots, but will also be able to utilize a live-viewing feature. Along with live-viewing, the ability to edit connections, including blocking them, will be available based on accessing an assigned ID to each connection. Similarly, other abilities will be provided to the user to export parts of the connections into a JSON snapshot, compatible with other applications. Overall, the application will be able to interface with other applications to provide connection information for the purposes of data processing or usage in other contexts.

A stable version of the application should be expected by January of 2027, although there may be interference that causes delays. It is my opinion that the program should be maintained after its creation to sustain continuous deployment principles, responding to customer input and adapting to it through architectural refactoring.

In the end, the project was a learning experience in many ways, allowing me to explore Python and its networking abilities. The simplistic beauty of Python combined with the near-endless expanse of computer networking technologies allows for exploration and improvements to be made to the program in the long term.