diff --git a/.github/BOTMETA.yml b/.github/BOTMETA.yml index 30135a4f94..8c212c65d4 100644 --- a/.github/BOTMETA.yml +++ b/.github/BOTMETA.yml @@ -241,6 +241,8 @@ files: $lookups/filetree.py: maintainers: dagwieers $lookups/flattened.py: {} + $lookups/github_app_access_token.py: + maintainers: weisheng-p $lookups/hiera.py: maintainers: jparrill $lookups/keyring.py: {} diff --git a/plugins/lookup/github_app_access_token.py b/plugins/lookup/github_app_access_token.py new file mode 100644 index 0000000000..265dacc504 --- /dev/null +++ b/plugins/lookup/github_app_access_token.py @@ -0,0 +1,156 @@ +# -*- coding: utf-8 -*- +# Copyright (c) 2023, Poh Wei Sheng +# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) +# SPDX-License-Identifier: GPL-3.0-or-later +from __future__ import (absolute_import, division, print_function) +__metaclass__ = type + +DOCUMENTATION = ''' + name: github_app_access_token + author: + - Poh Wei Sheng (@weisheng-p) + short_description: Obtain short-lived Github App Access tokens + version_added: '8.2.0' + requirements: + - jwt (https://github.com/GehirnInc/python-jwt) + description: + - This generates a Github access token that can be used with a C(git) command, if you use a Github App. + options: + key_path: + description: + - Path to your private key. + required: true + type: path + app_id: + description: + - Your GitHub App ID, you can find this in the Settings page. + required: true + type: str + installation_id: + description: + - The installation ID that contains the git repository you would like access to. + - As of 2023-12-24, this can be found via Settings page > Integrations > Application. The last part of the URL in the + configure button is the installation ID. + - Alternatively, you can use PyGithub (U(https://github.com/PyGithub/PyGithub)) to get your installation ID. + required: true + type: str + token_expiry: + description: + - How long the token should last for in seconds. + default: 600 + type: int +''' + +EXAMPLES = ''' +- name: Get access token to be used for git checkout with app_id=123456, installation_id=64209 + ansible.builtin.git: + repo: >- + https://x-access-token:{{ github_token }}@github.com/hidden_user/super-secret-repo.git + dest: /srv/checkout + vars: + github_token: >- + lookup('github_app_token', key_path='/home/to_your/key', + app_id='123456', installation_id='64209') +''' + +RETURN = ''' + _raw: + description: A one-element list containing your GitHub access token. + type: list + elements: str +''' + + +try: + from jwt import JWT, jwk_from_pem + HAS_JWT = True +except ImportError: + HAS_JWT = False + +import time +import json +from ansible.module_utils.urls import open_url +from ansible.module_utils.six.moves.urllib.error import HTTPError +from ansible.errors import AnsibleError +from ansible.plugins.lookup import LookupBase +from ansible.utils.display import Display + +if HAS_JWT: + jwt_instance = JWT() +else: + jwk_from_pem = None + jwt_instance = None + +display = Display() + + +def read_key(path): + try: + with open(path, 'rb') as pem_file: + return jwk_from_pem(pem_file.read()) + except Exception as e: + raise AnsibleError("Error while parsing key file: {0}".format(e)) + + +def encode_jwt(app_id, jwk, exp=600): + now = int(time.time()) + payload = { + 'iat': now, + 'exp': now + exp, + 'iss': app_id, + } + try: + return jwt_instance.encode(payload, jwk, alg='RS256') + except Exception as e: + raise AnsibleError("Error while encoding jwt: {0}".format(e)) + + +def post_request(generated_jwt, installation_id): + github_api_url = f'https://api.github.com/app/installations/{installation_id}/access_tokens' + headers = { + "Authorization": f'Bearer {generated_jwt}', + "Accept": "application/vnd.github.v3+json", + } + try: + response = open_url(github_api_url, headers=headers, method='POST') + except HTTPError as e: + try: + error_body = json.loads(e.read().decode()) + display.vvv("Error returned: {0}".format(error_body)) + except Exception: + error_body = {} + if e.code == 404: + raise AnsibleError("Github return error. Please confirm your installationd_id value is valid") + elif e.code == 401: + raise AnsibleError("Github return error. Please confirm your private key is valid") + raise AnsibleError("Unexpected data returned: {0} -- {1}".format(e, error_body)) + response_body = response.read() + try: + json_data = json.loads(response_body.decode('utf-8')) + except json.decoder.JSONDecodeError as e: + raise AnsibleError("Error while dencoding JSON respone from github: {0}".format(e)) + return json_data.get('token') + + +def get_token(key_path, app_id, installation_id, expiry=600): + jwk = read_key(key_path) + generated_jwt = encode_jwt(app_id, jwk, exp=expiry) + return post_request(generated_jwt, installation_id) + + +class LookupModule(LookupBase): + def run(self, terms, variables=None, **kwargs): + if not HAS_JWT: + raise AnsibleError('Python jwt library is required. ' + 'Please install using "pip install jwt"') + + self.set_options(var_options=variables, direct=kwargs) + + t = get_token( + self.get_option('key_path'), + self.get_option('app_id'), + self.get_option('installation_id'), + self.get_option('token_expiry'), + ) + + return [t] diff --git a/tests/unit/plugins/lookup/test_github_app_access_token.py b/tests/unit/plugins/lookup/test_github_app_access_token.py new file mode 100644 index 0000000000..4bf9c7e704 --- /dev/null +++ b/tests/unit/plugins/lookup/test_github_app_access_token.py @@ -0,0 +1,52 @@ +# -*- coding: utf-8 -*- +# Copyright (c) 2023, Poh Wei Sheng +# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt) +# SPDX-License-Identifier: GPL-3.0-or-later + +from __future__ import (absolute_import, division, print_function) +__metaclass__ = type +import json + +from ansible_collections.community.general.tests.unit.compat import unittest +from ansible_collections.community.general.tests.unit.compat.mock import ( + patch, + MagicMock, + mock_open +) +from ansible.plugins.loader import lookup_loader + + +class MockJWT(MagicMock): + def encode(self, payload, key, alg): + return 'Foobar' + + +class MockResponse(MagicMock): + response_token = 'Bar' + + def read(self): + return json.dumps({ + "token": self.response_token, + }).encode('utf-8') + + +class TestLookupModule(unittest.TestCase): + + def test_get_token(self): + with patch.multiple("ansible_collections.community.general.plugins.lookup.github_app_access_token", + open=mock_open(read_data="foo_bar"), + open_url=MagicMock(return_value=MockResponse()), + jwk_from_pem=MagicMock(return_value='private_key'), + jwt_instance=MockJWT(), + HAS_JWT=True): + lookup = lookup_loader.get('community.general.github_app_access_token') + self.assertListEqual( + [MockResponse.response_token], + lookup.run( + [], + key_path="key", + app_id="app_id", + installation_id="installation_id", + token_expiry=600 + ) + )