Secrets management, operational security, keeping API tokens hidden while streaming

Question

I've taken up code streaming and I'm concerned that as I do so, I'll leak an API token somewhere.

It was suggested that I use something like "a secrets.toml or secrets.yaml or secrets.json file." But I don't want it all in one file. If I did have a leak, I'd likely lose all my tokens at once.

So I considered a secrets directory, where each file holds each API's token. But what if I accidentally cat the file to the screen?

So I wanted to encrypt the tokens. Because I'm a bit obsessive about these kinds of things. I wouldn't call this encryption, but it is obfuscation, and I think it improves my op-sec quite a bit and it makes me feel a bit better about not leaking api tokens.

I'm using Python, and specifically the following functions and the Path object from the standard library. randbits will give me as close to cryptographically strong random 0s and 1s as one can get from the Python standard library. getpass will hide the token if I happen to paste it there. And Path objects are amazing for dealing with file paths.

from sys import argv
from secrets import randbits
from pathlib import Path
from getpass import getpass
from tempfile import mkdtemp


_USAGE = """usage:
  python -m py.token "api name"
  python -m py.token --test
"""

def main():
    if len(argv) == 2:
        if argv[1] == '--test':
            test()
        else:
            try:
                print(Manager().get_token(argv[1]))
            except:
                print(_USAGE)
                raise
    else:
        print(_USAGE)

As the main() foreshadows, to manage it all I have a Manager class that can save the tokens and update them, update all of them with a new mask, and for regular usage, to get api tokens:

class Manager:
    def __init__(self, root=Path.home()):
        self.passtoken_dir = root / 'pass'
        self.secrets_dir = self.passtoken_dir / 'secrets'
        self.ptfile = self.passtoken_dir / 'token'
        self.create_dirs_and_mask()

    def create_dirs_and_mask(self):
        self.passtoken_dir.mkdir(exist_ok=True)
        self.secrets_dir.mkdir(exist_ok=True)
        if not self.ptfile.is_file():
            self.ptfile.write_bytes(new_passtoken())
        else:
            # just realized this print() won't work for command line usage.
            print('we have a mask file already.')

    def save_token(self, api_name='', token=b''):
        if not api_name:
            api_name = input('api name: ')
        if not token:
            token = bytes(getpass('input token (hidden): '), 'utf8')
        file = self.secrets_dir / api_name
        file.write_bytes(mask(token, self.ptfile.read_bytes()))

    def update_mask_and_masked_tokens(self):
        old_pt = self.ptfile.read_bytes()
        new_pt = new_passtoken()
        for file in self.secrets_dir.iterdir():
            token = unmask(file.read_bytes(), old_pt)
            file.write_bytes(mask(token, new_pt))
        self.ptfile.write_bytes(new_pt)

    def get_token(self, api_name) -> bytes:
        return unmask((self.secrets_dir/api_name).read_bytes(),
                      self.ptfile.read_bytes())

I like having the create_dirs_and_mask method separate for the possibility of overriding for test purposes, but I didn't go in that direction for testing.

The above Manager object relies heavily on these utility functions to obfuscate and convert from bytes to ints and back.

def new_passtoken() -> bytes:
    return int_to_bytes(randbits(8*256))

def bytes_to_int(token) -> int:
    return int.from_bytes(token, 'big')

def int_to_bytes(integer) -> bytes:
    # wish we didn't have to implement ourselves...
    result = []
    while integer:
        result.append(integer & 255)
        integer >>= 8
    return bytes(reversed(result))

def mask(token: bytes, passtoken: bytes) -> bytes:
    return int_to_bytes(bytes_to_int(token)
                        ^ bytes_to_int(passtoken))

def unmask(masked_token: bytes, passtoken: bytes) -> bytes:
    return int_to_bytes(bytes_to_int(masked_token)
                        ^ bytes_to_int(passtoken))

I first implemented mask and unmask with multiply and integer division instead of xor, but it seemed to me that if a leak happened there would be a greater chance of inferring the components of the calculation.

This is all one module, and I haven't yet put the tests in a test module. Tests do pass:

def test(): # TODO put tests in a test package
    """test that functions round-trip
    and manager can save, get, update, and still get.
    """
    from shutil import rmtree
    token = b'abc123}|{'
    assert token == int_to_bytes(bytes_to_int(token))
    passtoken = b'passtoken'
    assert token == unmask(mask(token, passtoken), passtoken)
    root = Path(mkdtemp()) 
    try:
        manager = Manager(root=root)
        api = 'any api'
        manager.save_token(api, token)
        assert manager.get_token(api) == token
        manager.update_mask_and_masked_tokens()
        assert manager.get_token(api) == token
        globals().update(locals())
    finally:
        rmtree(root)

For the command line usage, we call the main when we're the entry point for the program:

if __name__ == '__main__':
    main()

Please review! I'm especially looking for suggestions that upgrade my attempts at "encryption" without putting a reliance on a third-party library, unless it is quite stable, seasoned, and expected to continue to be in service for the foreseeable future.

Answer 1

1. Your main is a classic arrow anti-pattern. You can use guard clauses to make the code flat and easier to understand.
2. Bare excepts are normally not a good idea. Why would you want to print the usage if the code works fine but the user passes a keyboard interrupt? You may want to use except Exception:.

3. I would prefer if mask and unmask were wrapped up in a class. I personally would leave them as static methods.

   This has the benefit that your Manager can be later changed to use any encryption. As long as there's an object that defines an encrypt and a decrypt method.

4. Functions like create_dirs_and_mask are normally a really large red flag, to me, signalling that your constructor is doing too much.

   Your constructor is doing quite a lot, building three paths from root, building two directories, building a file if it doesn't exist, printing to the end user if a file already exists.

   Not only does this clearly break SRP, it's needlessly locking down customizability of the class and making it more annoying to test.

   Just move the fancy bits into a class method and leave the constructor to be as dumb as possible.

5. The name update_mask_and_masked_tokens is a mouthful and a half. I would just change this to be the setter of the token property.

   Since the function currently doesn't take an argument I would move the new_passtoken() call out of the method and as an argument.

6. You have a confusing and useless line masked = mask.

7. I would change Manager to act like a dict - changing save_token to __setitem__.

   Changing it to have the same interface as a dict allows for you to easily swap this out for a plain old dictionary to easily test code that uses this. It also ensures that you don't break SRP by adding bells or whistles that don't belong on Manager. For example save_token is currently merging business logic with its user interface, a generally pretty poor design choice.

8. The special bells and whistles added to Manager.save_token are currently not used, you may want to follow YAGNI and just scrap that part of your code. Alternately you can expose the functionality as an external function.
9. I'm not a fan of python -m py.token --test. I personally would just drop it as an option and use pytest.
10. Rather than mkdtemp I would use TemporaryDirectory and use it as a context manager. This would remove the need to import shutil.rmtree.

from sys import argv
from secrets import randbits
from pathlib import Path
from getpass import getpass
from tempfile import TemporaryDirectory

_USAGE = """usage:
  python -m py.token "api name"
  python -m py.token --test
"""


def main():
    if len(argv) != 2:
        print(_USAGE)
        return

    if argv[1] == '--test':
        test()
        return

    try:
        manager = TokenManager.load(new_passtoken)
        print(manager[argv[1]])
    except Exception:
        print(_USAGE)
        raise


class Mask:
    @staticmethod
    def encrypt(token: bytes, passtoken: bytes) -> bytes:
        return int_to_bytes(bytes_to_int(token)
                            ^ bytes_to_int(passtoken))

    @staticmethod
    def decrypt(masked_token: bytes, passtoken: bytes) -> bytes:
        return int_to_bytes(bytes_to_int(masked_token)
                            ^ bytes_to_int(passtoken))


class TokenManager:
    def __init__(self, token_file, secrets_dir, crypto):
        secrets_dir.mkdir(exist_ok=True)
        self._secrets_dir = secrets_dir
        self._token_file = token_file
        self._token = token_file.read_bytes()
        self._crypto = crypto

    @classmethod
    def load(cls, new_token=None, crypto=Mask, root=Path.home()):
        base = root / 'pass'
        base.mkdir(exist_ok=True)
        secrets = base / 'secrets'
        token = base / 'token'
        if not token.exists():
            if new_token is None:
                raise ValueError('No existing token exists')
            token.write_bytes(new_token())
        return cls(token, secrets, crypto)

    def __getitem__(self, api) -> bytes:
        encrypted = (self._secrets_dir / api).read_bytes()
        return self._crypto.decrypt(encrypted, self.token)

    def __setitem__(self, api, token) -> None:
        encrypted = self._crypto.encrypt(token, self.token)
        (self._secrets_dir / api).write_bytes(encrypted)

    @property
    def token(self) -> bytes:
        return self._token

    @token.setter
    def token(self, token) -> None:
        for file in self._secrets_dir.iterdir():
            token_ = self._crypto.decrypt(file.read_bytes(), self.token)
            encrypted = self._crypto.encrypt(token_, token)
            file.write_bytes(encrypted)
        self._token = token
        self._token_file.write_bytes(token)


def new_passtoken() -> bytes:
    return int_to_bytes(randbits(8*256))


def bytes_to_int(token) -> int:
    return int.from_bytes(token, 'big')


def int_to_bytes(integer) -> bytes:
    # wish we didn't have to implement ourselves...
    result = []
    while integer:
        result.append(integer & 255)
        integer >>= 8
    return bytes(reversed(result))


def save_token(manager, api_name='', token=b''):
    if not api_name:
        api_name = input('api name: ')
    if not token:
        token = bytes(getpass('input token (hidden): '), 'utf8')
    manager[api_name] = token


def test():
    token = b'abc123}|{'
    assert token == int_to_bytes(bytes_to_int(token))
    passtoken = b'passtoken'
    assert token == Mask.decrypt(Mask.encrypt(token, passtoken), passtoken)
    with TemporaryDirectory() as tmp_dir:
        manager = TokenManager.load(new_passtoken, crypto=Mask, root=Path(tmp_dir))
        api = 'any api'
        manager[api] = token
        assert manager[api] == token
        manager.token = new_passtoken()
        assert manager[api] == token
        globals().update(locals())


if __name__ == '__main__':
    main()

Answer 2

def int_to_bytes(integer) -> bytes:
    # wish we didn't have to implement ourselves...
    result = []
    while integer:
        result.append(integer & 255)
        integer >>= 8
    return bytes(reversed(result))

Your wish has been granted; you don't have to implement it yourself:

def int_to_bytes(integer) -> bytes:
    return integer.to_bytes((integer.bit_length() + 7) // 8, 'big')

Note: int.bit_length() is the length of the value in bits. We must divide this by 8 to yield the required number of bytes, but we must not lose any fractions of a byte, so we add 7 to the number of bits first. For example: If a number requires 81 bits, (10.125 bytes, or 1 bit more than 10 bytes), adding 7 will increase this to 88 bits, which integer-divided by 8 gives 11, as required. int.to_bytes(num_bytes, 'big') will then result in the correct bytes result.

Answer 3

You also have an operational problem as much as a code problem. Any tokens you use for a public demonstration should -

1. Be as least-privileged as possible (eg grant READ access instead of CRUD or full admin)
2. Be tied to demo/test/dev environments where possible (I pray you are not doing code streaming in PROD)
3. Expired/rotated/destroyed frequently and aggressively (Terminate the token when your stream is done, set shorter timeouts so it expires for you)
4. Audited for improper use (check your logs and set alerts for unexpected use)
5. Access controlled in addition to the token where possible (whitelist IP, mutual TLS, etc).

The code is an interesting exercise and the reviews will help. However what prevents some other debugging operation from exposing a token in a header or log or step-wise debugging operation? You can expose a token in a log or tcpdump just as easily as with cat.