Ansiblie lookup plugin for Yandex Cloud lockbox

yandex_lockbox.py

# Usage: '{{ lookup("yandex_lockbox", "e6r3ox0h4q0", field="some_secret") }}'
import os
import os.path
import time
import json
import urllib.request
import jwt
from ansible.plugins.lookup import LookupBase
from ansible.errors import AnsibleError

class LookupModule(LookupBase):

    def _get_yc_iam_token_from_service_account_key(self, key_path):
        now = int(time.time())
        with open(os.path.expanduser(key_path), 'r') as key_file:
            key = json.load(key_file)

        payload = {
            'aud': 'https://iam.api.cloud.yandex.net/iam/v1/tokens',
            'iss': key["service_account_id"],
            'iat': now,
            'exp': now + 1800,
        }
        jwt_token = jwt.encode(
            payload,
            key["private_key"],
            algorithm='PS256',
            headers={'kid': key["id"]}
        )

        request = urllib.request.Request("https://iam.api.cloud.yandex.net/iam/v1/tokens", json.dumps({
            "jwt": jwt_token,
        }).encode("utf-8"))
        with urllib.request.urlopen(request) as response:
            data = json.load(response)

        return data['iamToken']

    def _get_yc_iam_token(self, oauth_token):
        request = urllib.request.Request("https://iam.api.cloud.yandex.net/iam/v1/tokens", json.dumps({
            "yandexPassportOauthToken": oauth_token,
        }).encode("utf-8"))
        with urllib.request.urlopen(request) as response:
            data = json.load(response)
        return data['iamToken']

    def run(self, keys, field=None, **kwargs):
        if len(keys) != 1:
            raise AnsibleError("exactly one positional argument required for yandex_lockbox lookup")
        key = keys[0]
        if os.environ.get('YC_KEY_PATH'):
            iam_token = self._get_yc_iam_token_from_service_account_key(os.environ['YC_KEY_PATH'])
        elif os.environ.get('YC_TOKEN'):
            iam_token = self._get_yc_iam_token(os.environ['YC_TOKEN'])
        else:
            raise AnsibleError("Not YC_TOKEN nor YC_KEY_PATH provided")

        if field is None:
            raise AnsibleError("field argument required for yandex_lockbox lookup")

        payload_url = "https://payload.lockbox.api.cloud.yandex.net/lockbox/v1/secrets/%s/payload" % key
        request = urllib.request.Request(payload_url, headers={"Authorization": "Bearer %s" % iam_token})
        with urllib.request.urlopen(request) as response:
            data = json.load(response)

        for entry in data['entries']:
            if entry['key'] == field:
                return [entry["textValue"]]

        raise AnsibleError("field %s not found")

Привет, оно работает? и как обстоят дела с файлов который надо получить из lockbox. Я не программист и мне пока не ясно как пройти авторизацию чтобы получить секреты, не подскажешь?

Привет, нужно получить токен вот по этой инструкции https://cloud.yandex.ru/docs/iam/concepts/authorization/oauth-token, записать его в переменные окружения через export YC_TOKEN=xxxx, и потом в этом же шелле запускать ансибл как обычно.

Файлы в текущем состоянии оно не умеет вытягивать, только строки. По идее файлы должно быть не сложно добавить, нужно смотреть в документацию по апи локбокса, не стал делать потому что было не нужно, секреты обычно небольшие, можно строкой в поле их уместить.