[PR #7761/33180344 backport][stable-8] Add github_app_access_token lookup plugin (#7802)

Add github_app_access_token lookup plugin (#7761)

* Add github_app_access_token lookup plugin

* Fix a typo in short_description

* Remove unused MockOpenUrl

* Fix MockJWT to be used on jwt_instance instead

* Fix a bunch of pep8 and pylint issue

* Remove JWT from requirements, also default jwt_instance and jwk_from_pem so they can be mocked

* Update version added

Co-authored-by: Felix Fontein <felix@fontein.de>

* Update git reference in doc

Co-authored-by: Felix Fontein <felix@fontein.de>

* Update plugins/lookup/github_app_access_token.py

Co-authored-by: Felix Fontein <felix@fontein.de>

* Expose token expiry as a configurable option

* Update BOTMETA.yml

* Update documentation

* Update example with var, so it is more readable

Co-authored-by: Felix Fontein <felix@fontein.de>

* Apply suggestions from code review

Co-authored-by: Felix Fontein <felix@fontein.de>

---------

Co-authored-by: Felix Fontein <felix@fontein.de>
(cherry picked from commit 3318034403)

Co-authored-by: weisheng-p <weisheng-p@users.noreply.github.com>

.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: {}

plugins/lookup/github_app_access_token.py

@@ -0,0 +1,156 @@
+# -*- coding: utf-8 -*-
+# Copyright (c) 2023, Poh Wei Sheng <weisheng-p@hotmail.sg>
+# 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]

tests/unit/plugins/lookup/test_github_app_access_token.py

@@ -0,0 +1,52 @@
+# -*- coding: utf-8 -*-
+# Copyright (c) 2023, Poh Wei Sheng <weisheng-p@hotmail.sg>
+# 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
+                )
+            )