test_socket.py

#!/usr/bin/env python3
# This was a failed PoC to try and use socket files as ssh_config file. :)
# by Stefan Midjich

import os
import functools
import asyncio


def set_socket_permission(socket_file):
    os.chmod(socket_file, 0o600)


class SSHConfigServer(asyncio.Protocol):

    def __init__(self):
        self._configs = [{
            'Host': 'syslog03',
            'HostName': '127.0.0.1',
            'User': 'stemid',
            'Port': 22
        }]


    def connection_made(self, transport):
        for config in self._configs:
            # Write first line of host config and delete that key from
            # local dict so we can loop out the rest.
            transport.write(bytes(config['Host'], encoding='UTF-8'))
            del config['Host']

            for (key, value) in config.items():
                _out = '{key} {value}\n'.format(
                    key=key,
                    value=value
                )
                transport.write(bytes(_out, encoding='UTF-8'))

def main():
    socket_filepath = "/Users/stemid/.ssh/dynamic_config"
    loop = asyncio.get_event_loop()
    server_coroutine = loop.create_unix_server(SSHConfigServer, socket_filepath)
    server = loop.run_until_complete(server_coroutine)

    callback = functools.partial(set_socket_permission, socket_filepath)
    loop.call_soon(callback)

    try:
        loop.run_forever()
    except KeyboardInterrupt:
        print("exit")
    finally:
        server.close()
        loop.close()
        os.remove(socket_filepath)


if __name__ == "__main__":
    main()