1
\$\begingroup\$

A small Python lib to renew key from Vault every 30s to always get the latest key.

from __future__ import print_function

import hvac
import time
import threading
import os


VAULT_URL = os.environ['VAULT_ADDR']
VAULT_TOKEN = os.environ['VAULT_TOKEN']


class Client:
    def __init__(self, *keys):
        self.keys = keys
        self.data_dict = {}
        self.client = hvac.Client(
            url=VAULT_URL,
            token=VAULT_TOKEN)

        self.__renew()

    def read(self):
        for key in self.keys:
            self.data_dict[key] = self.client.read(key)
        return self.data_dict

    def __renew(self):
        self.client.renew_token()
        threading.Timer(30, self.__renew).start()
        self.read()

if __name__ == '__main__':
    client = Client('secret/key')
    print(client.read())
\$\endgroup\$
  • \$\begingroup\$ What kind of output does this generate? A steady stream of new keys, a onetime output? \$\endgroup\$ – holroy May 12 '17 at 20:05
1
\$\begingroup\$

I don't really see the need for threading here, as it seems like it only generates a one-time key. If this was some kind of long-time running process it would make a little more sense, but I still believe I would rather use a method which kept track of the last request, and if that request was made more than 30 seconds ago then read a new key from the hvac.Client.

The following code is totally untested:

import hvac
import time
import threading
import os
import datetime

VAULT_URL = os.environ['VAULT_ADDR']
VAULT_TOKEN = os.environ['VAULT_TOKEN']

class Client:
    def __init__(self, *keys):
        self.keys = keys
        self.data_dict = {}
        self.client = hvac.Client(
            url=VAULT_URL,
            token=VAULT_TOKEN)

        self.read_keys()


    def read_keys(self):
        self.client.renew_token()

        for key in self.keys:
            self.data_dict[key] = self.client.read(key)

        self.last_read = datetime.now()


    def read(self):

        # If more than 30s since last read, renew keys
        if (datetime.now() - self.last_read).total_seconds() > 30:
            self.read_keys()

        # Return the secret keys
        return self.data_dict

Hopefully, not too many errors in there, but it should the gist of the idea, if I understood your underlying requirements of getting new keys every 30s or so correctly.

\$\endgroup\$
  • \$\begingroup\$ Can you maybe explain a bit more why threading is not needed? \$\endgroup\$ – toy May 17 '17 at 22:02
1
\$\begingroup\$

You don't need a class here. Neither you need a Timer.

import hvac
import time
import threading
import os
import sys


VAULT_URL, VAULT_TOKEN = sys.argv[1:]


def read(client, *keys):
    client.renew_token()
    return {k: client.read(key) for key in keys}


if __name__ == '__main__':
    client = hvac.Client(url=VAULT_URL, token=VAULT_TOKEN)
    while True:
        print(read(client, 'secret/key'))
        time.sleep(30)
\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.